Pyspark - объединяет все столбцы данных

Question

Я хочу сгруппировать фрейм данных в один столбец, а затем применить статистическую функцию ко всем столбцам.

Например, у меня есть df с 10 столбцами. Я хочу сгруппировать по первому столбцу «1», а затем применить статистическую функцию «сумма» ко всем оставшимся столбцам (все они числовые).

Эквивалентом R этого является summarise_all. Ex в R.

df = df%>%group_by(column_one)%>%summarise_all(funs(sum))

Я не хочу вручную вводить столбцы в статистической команде в pyspark, так как количество столбцов в кадре данных будет динамическим.

Answer 1

Вот пример для pyspark:

df = spark.createDataFrame([(1,10,20,30),(1,10,20,30),(1,10,20,30)],['id','value1','value2','value3'])

exprs = {x: "sum" for x in df.columns if x is not df.columns[0]}

df.groupBy("id").agg(exprs).show()

#+---+-----------+-----------+-----------+
#| id|sum(value2)|sum(value1)|sum(value3)|
#+---+-----------+-----------+-----------+
#|  1|         60|         30|         90|
#+---+-----------+-----------+-----------+

Здесь df.columns[0] представляет первый столбец df.Вы также можете указать столбец groupBy по имени, как показано ниже.

exprs = {x: "sum" for x in df.columns if x is not 'id'}

Answer 2

Если бы я делал это в spark scala, я бы использовал имена столбцов и определял функции агрегирования следующим образом:

val df = List(("a", 1,2,3), ("a", 4,4,4)).toDF("id", "a", "b", "c")

// Note: df.columns.tail returns Array(a, b, c)
val aggs = df.columns.tail.map(_ -> "sum").toMap
//aggs: scala.collection.immutable.Map[String,String] = Map(a -> sum, b -> sum, c -> sum)

// Group by and execute aggregates:
df.groupBy($"id").agg(aggs).show
+---+------+------+------+
| id|sum(a)|sum(b)|sum(c)|
+---+------+------+------+
|  a|     5|     6|     7|
+---+------+------+------+

Другой вариант - запускать одинаковые sum для всех указанных имен столбцов:

df.groupBy($"id").sum(df.columns.tail: _*).show()  // to python users, :_* is a scala operator used to expand a list into a vararg

ПРИМЕЧАНИЕ: вас может заинтересовать эта документация: https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.sql.RelationalGroupedDataset