Spark SQL: apply aggregate functions to a list of columns
Czy istnieje sposób na zastosowanie funkcji zbiorczej do wszystkich (lub listy) kolumn ramki danych, wykonując groupBy
? Innymi słowy, czy istnieje sposób, aby uniknąć tego dla każdej kolumny:
df.groupBy("col1")
.agg(sum("col2").alias("col2"), sum("col3").alias("col3"), ...)
3 answers
Istnieje wiele sposobów zastosowania funkcji agregowanych do wielu kolumn.
GroupedData
klasa zapewnia szereg metod dla najbardziej popularnych funkcji, w tym count
, max
, min
, mean
i sum
, które mogą być używane bezpośrednio w następujący sposób:
-
Python:
df = sqlContext.createDataFrame( [(1.0, 0.3, 1.0), (1.0, 0.5, 0.0), (-1.0, 0.6, 0.5), (-1.0, 5.6, 0.2)], ("col1", "col2", "col3")) df.groupBy("col1").sum() ## +----+---------+-----------------+---------+ ## |col1|sum(col1)| sum(col2)|sum(col3)| ## +----+---------+-----------------+---------+ ## | 1.0| 2.0| 0.8| 1.0| ## |-1.0| -2.0|6.199999999999999| 0.7| ## +----+---------+-----------------+---------+
-
Scala
val df = sc.parallelize(Seq( (1.0, 0.3, 1.0), (1.0, 0.5, 0.0), (-1.0, 0.6, 0.5), (-1.0, 5.6, 0.2)) ).toDF("col1", "col2", "col3") df.groupBy($"col1").min().show // +----+---------+---------+---------+ // |col1|min(col1)|min(col2)|min(col3)| // +----+---------+---------+---------+ // | 1.0| 1.0| 0.3| 0.0| // |-1.0| -1.0| 0.6| 0.2| // +----+---------+---------+---------+
Opcjonalnie można przekazać listę kolumn, które powinny być zagregowane
df.groupBy("col1").sum("col2", "col3")
Można również przekazać słownik / mapę z kolumny a klucze i funkcje jako wartości:
-
Python
exprs = {x: "sum" for x in df.columns} df.groupBy("col1").agg(exprs).show() ## +----+---------+ ## |col1|avg(col3)| ## +----+---------+ ## | 1.0| 0.5| ## |-1.0| 0.35| ## +----+---------+
-
Scala
val exprs = df.columns.map((_ -> "mean")).toMap df.groupBy($"col1").agg(exprs).show() // +----+---------+------------------+---------+ // |col1|avg(col1)| avg(col2)|avg(col3)| // +----+---------+------------------+---------+ // | 1.0| 1.0| 0.4| 0.5| // |-1.0| -1.0|3.0999999999999996| 0.35| // +----+---------+------------------+---------+
Wreszcie możesz użyć varargs:
-
Python
from pyspark.sql.functions import min exprs = [min(x) for x in df.columns] df.groupBy("col1").agg(*exprs).show()
-
Scala
import org.apache.spark.sql.functions.sum val exprs = df.columns.map(sum(_)) df.groupBy($"col1").agg(exprs.head, exprs.tail: _*)
Istnieją inne sposoby osiągnięcia podobnego efektu, ale te powinny być bardziej niż wystarczające przez większość czasu.
Zobacz też:
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/doraprojects.net/template/agent.layouts/content.php on line 54
2019-03-07 20:22:27
Kolejny przykład tego samego pojęcia - ale powiedzmy - masz 2 różne kolumny - i chcesz zastosować różne funkcje agg do każdej z nich tj.]}
f.groupBy("col1").agg(sum("col2").alias("col2"), avg("col3").alias("col3"), ...)
Oto sposób, aby to osiągnąć - choć jeszcze nie wiem, jak dodać alias w tym przypadku
Zobacz przykład poniżej-Korzystanie z map
val Claim1 = StructType(Seq(StructField("pid", StringType, true),StructField("diag1", StringType, true),StructField("diag2", StringType, true), StructField("allowed", IntegerType, true), StructField("allowed1", IntegerType, true)))
val claimsData1 = Seq(("PID1", "diag1", "diag2", 100, 200), ("PID1", "diag2", "diag3", 300, 600), ("PID1", "diag1", "diag5", 340, 680), ("PID2", "diag3", "diag4", 245, 490), ("PID2", "diag2", "diag1", 124, 248))
val claimRDD1 = sc.parallelize(claimsData1)
val claimRDDRow1 = claimRDD1.map(p => Row(p._1, p._2, p._3, p._4, p._5))
val claimRDD2DF1 = sqlContext.createDataFrame(claimRDDRow1, Claim1)
val l = List("allowed", "allowed1")
val exprs = l.map((_ -> "sum")).toMap
claimRDD2DF1.groupBy("pid").agg(exprs) show false
val exprs = Map("allowed" -> "sum", "allowed1" -> "avg")
claimRDD2DF1.groupBy("pid").agg(exprs) show false
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/doraprojects.net/template/agent.layouts/content.php on line 54
2016-11-14 18:38:14
Aktualne odpowiedzi są całkowicie poprawne, jak utworzyć agregacje, ale żadna nie odnosi się do aliasu/zmiany nazwy kolumny, która jest również wymagana w pytaniu.
Zazwyczaj tak się zajmuję tą sprawą:
val dimensionFields = List("col1")
val metrics = List("col2", "col3", "col4")
val columnOfInterests = dimensions ++ metrics
val df = spark.read.table("some_table").
.select(columnOfInterests.map(c => col(c)):_*)
.groupBy(dimensions.map(d => col(d)): _*)
.agg(metrics.map( m => m -> "sum").toMap)
.toDF(columnOfInterests:_*) // that's the interesting part
Ostatni wiersz zasadniczo zmienia nazwy każdej kolumny zagregowanej ramki danych na oryginalne pola, zasadniczo zmieniając sum(col2)
i sum(col3)
Na po prostu col2
i col3
.
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/doraprojects.net/template/agent.layouts/content.php on line 54
2020-05-08 15:55:00