有没有办法在执行 groupBy 时将聚合函数应用于数据帧的所有(或列表)列?换句话说,有没有办法避免为每一列执行此操作:
groupBy
df.groupBy("col1") .agg(sum("col2").alias("col2"), sum("col3").alias("col3"), ...)
同一概念的另一个例子 - 但是说 - 你有两个不同的列 - 并且你想对它们中的每一个应用不同的agg函数,即
f.groupBy("col1").agg(sum("col2").alias("col2"), avg("col3").alias("col3"), ...)
这是实现它的方法 - 虽然我还不知道在这种情况下如何添加别名
请参阅下面的示例 - 使用 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
将聚合函数应用于多个列有多种方法 .
GroupedData 类为最常用的函数提供了许多方法,包括 count , max , min , mean 和 sum ,可以直接使用如下:
GroupedData
count
max
min
mean
sum
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| ## +----+---------+-----------------+---------+
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| // +----+---------+---------+---------+
您可以选择传递应该聚合的列的列表
df.groupBy("col1").sum("col2", "col3")
您还可以使用键和字符串传递字典/映射作为值:
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| ## +----+---------+
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| // +----+---------+------------------+---------+
最后你可以使用varargs:
from pyspark.sql.functions import min exprs = [min(x) for x in df.columns] df.groupBy("col1").agg(*exprs).show()
import org.apache.spark.sql.functions.sum val exprs = df.columns.map(sum(_)) df.groupBy($"col1").agg(exprs.head, exprs.tail: _*)
还有其他一些方法可以达到类似的效果,但这些方法在大多数情况下应该足够了 .
2 回答
同一概念的另一个例子 - 但是说 - 你有两个不同的列 - 并且你想对它们中的每一个应用不同的agg函数,即
这是实现它的方法 - 虽然我还不知道在这种情况下如何添加别名
请参阅下面的示例 - 使用 Map
将聚合函数应用于多个列有多种方法 .
GroupedData
类为最常用的函数提供了许多方法,包括count
,max
,min
,mean
和sum
,可以直接使用如下:您可以选择传递应该聚合的列的列表
您还可以使用键和字符串传递字典/映射作为值:
最后你可以使用varargs:
还有其他一些方法可以达到类似的效果,但这些方法在大多数情况下应该足够了 .