Совокупность фреймов данных Spark по нескольким столбцам

Actually I am working on pyspark code. My dataframe is

+-------+--------+--------+--------+--------+
|element|collect1|collect2|collect3|collect4|
+-------+--------+--------+--------+--------+
|A1     |   1.02 |    2.6 |   5.21 |    3.6 |
|A2     |   1.61 |   2.42 |   4.88 |   6.08 |
|B1     |   1.66 |   2.01 |    5.0 |    4.3 |
|C2     |   2.01 |   1.85 |   3.42 |   4.44 |
+-------+--------+--------+--------+--------+

Мне нужно найти среднее значение и стандартное отклонение для каждого элемента, объединив все столбцы collectX. Окончательный результат должен быть таким, как показано ниже.

+-------+--------+--------+
|element|mean    |stddev  |
+-------+--------+--------+
|A1     |   3.11 |   1.76 |
|A2     |   3.75 |   2.09 |
|B1     |   3.24 |   1.66 |
|C2     |   2.93 |   1.23 |
+-------+--------+--------+

В приведенном ниже коде все средние значения в отдельных столбцах df.groupBy("element").mean().show(). Вместо того, чтобы делать это для каждого столбца, можно ли объединить все столбцы?

+-------+-------------+-------------+-------------+-------------+
|element|avg(collect1)|avg(collect2)|avg(collect3)|avg(collect4)|
+-------+-------------+-------------+-------------+-------------+
|A1     |   1.02      |   2.6       |   5.21      |    3.6      |
|A2     |   1.61      |   2.42      |   4.88      |   6.08      |
|B1     |   1.66      |   2.01      |    5.0      |    4.3      |
|C2     |   2.01      |   1.85      |   3.42      |   4.44      |
+-------+-------------+-------------+-------------+-------------+

Я попытался использовать функцию описания, поскольку она имеет полные функции агрегации, но все еще отображается как отдельный столбец df.groupBy("element").mean().describe().show()

Благодарность


pyspark apache-spark-sql spark-dataframe
person Chn    schedule 24.03.2016    source источник

Ответы (2)


arrow_upward
1
arrow_downward

Spark позволяет собирать все виды статистики по столбцам. Вы пытаетесь рассчитать статистику по строке. В этом случае вы можете взломать что-нибудь с помощью udf. Вот пример :D

$ pyspark
>>> from pyspark.sql.types import DoubleType
>>> from pyspark.sql.functions import array, udf
>>>
>>> mean = udf(lambda v: sum(v) / len(v), DoubleType())
>>> df = sc.parallelize([['A1', 1.02, 2.6, 5.21, 3.6], ['A2', 1.61, 2.42, 4.88, 6.08]]).toDF(['element', 'collect1', 'collect2', 'collect3', 'collect4'])
>>> df.show()
+-------+--------+--------+--------+--------+
|element|collect1|collect2|collect3|collect4|
+-------+--------+--------+--------+--------+
|     A1|    1.02|     2.6|    5.21|     3.6|
|     A2|    1.61|    2.42|    4.88|    6.08|
+-------+--------+--------+--------+--------+
>>> df.select('element', mean(array(df.columns[1:])).alias('mean')).show()
+-------+------+
|element|  mean|
+-------+------+
|     A1|3.1075|
|     A2|3.7475|
+-------+------+
person the.malkolm    schedule 24.03.2016

arrow_upward
0
arrow_downward

Вы пытались просто сложить столбцы вместе и, возможно, разделить на 4?

SELECT avg((collect1 + collect2 + collect3 + collect4) / 4),
  stddev((collect1 + collect2 + collect3 + collect4) / 4)

Это не будет делать именно то, что вы хотите, но поймите идею.

Не уверен, что ваш язык, но вы всегда можете построить запрос на лету, если вы не довольны жестко закодированным:

val collectColumns = df.columns.filter(_.startsWith("collect"))
val stmnt = "SELECT avg((" + collectColumns.mkString(" + ") + ") / " + collectColumns.length + "))"

Вы поняли идею.

person David Griffin    schedule 24.03.2016
comment
Фактически, фрейм данных может иметь больше или меньше столбцов collectX. Выполнение жесткого кода не является предпочтительным выбором. - person Chn; 24.03.2016