Spark SQL：将聚合函数应用于列列表

Question

在做分组 (groupBy) 时，是否有一种方法可以将聚合函数应用于数据帧的所有 (或一列) 列？换句话说，是否有一种避免为每个列都执行此操作的方法：

df.groupBy("col1")
  .agg(sum("col2").alias("col2"), sum("col3").alias("col3"), ...)

- lilloraffa

Answer 1

有多种方法可以将聚合函数应用于多个列。

GroupedData 类提供了许多常见函数的方法，包括 count、max、min、mean 和 sum，可以直接使用如下：

• 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|
// +----+---------+---------+---------+

可选地，您可以传递一个应该进行聚合的列列表

df.groupBy("col1").sum("col2", "col3")

您也可以传递字典/映射，其中列为键，函数为值：

• 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|
// +----+---------+------------------+---------+

最后你可以使用可变参数：

• 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: _*)

还有其他方法可以实现类似的效果，但大多数情况下这些方法已经足够了。

另请参阅：

• spark数据框架中对同一列进行多个聚合操作

- zero323

---

@javadba 不，这只意味着 Dataset.groupBy / Dataset.groupByKey 和 RDD.groupBy / RDD.groupByKey 在一般情况下具有不同的语义。在简单的 DataFrame 聚合中 查看此处。还有更多内容，但这里并不重要。 - zero323

谢谢。这是另一个有用的资源（主观上，自我推销警告）：https://git.io/vM1Ch - zero323

如何为列添加别名？ - GeekFactory

@GeekFactory exprs = [df.columns中的min(x).alias("{0}".format(x)) for x in df.columns] - zero323

@zero323 很棒的回答。如果我想为每列计算多个统计数据，应该怎么做呢？是这样的 exprs = [ [min(x).alias("{0}_min".format(x)), max(x).alias("{0}_max".format(x)), mean(x).alias("{0}_mean".format(x)) ] for x in df.columns] - Kenny

Answer 2

同一概念的另一个例子——假设您有两个不同的列，您想对每个列应用不同的聚合函数，即

f.groupBy("col1").agg(sum("col2").alias("col2"), avg("col3").alias("col3"), ...)

以下是实现它的方法 - 虽然我还不知道如何在这种情况下添加别名

请参考下面的示例 - 使用地图

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

- Sumit Pal

Answer 3

目前的答案完全正确，可以创建聚合，但实际上没有回答问题中还要求的列别名/重命名的问题。

通常，我会这样处理此类情况：

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

最后一行基本上将聚合数据框的每个列重命名为原始字段，从本质上改变了sum(col2)和sum(col3)为简单的col2和col3。

- Philippe Oger

Answer 4

对于Pyspark，您可以使用以下代码，它可以将不同的列名和聚合函数组合起来，并进行重命名。

from pyspark.sql.functions import expr
dims = ["col1"]
mertrics = ["col2", "col3", "col4"]
aggFuncs = ["sum", "avg", "max"]

aggs = [expr(f"{a}({c}) as {c}_{a}") for c in metrics for a in aggFuncs]
# or you can also do below, they are equivalent
# aggs = [expr(f"{a}({c})").alias(f"{c}_{a}") for c in metrics for a in aggFuncs]

df.groupBy(*dims).agg(*aggs).show()

- Gary Liu