加快 spark 数据帧 groupBy答案

【问题标题】：Speed up spark dataframe groupBy加快 spark 数据帧 groupBy
【发布时间】：2018-11-06 22:44:36
【问题描述】：

我在 Spark 方面相当缺乏经验，需要有关 groupBy 和数据帧上的聚合函数的帮助。考虑以下数据框：

val df = (Seq((1, "a", "1"),
              (1,"b", "3"),
              (1,"c", "6"),
              (2, "a", "9"),
              (2,"c", "10"),
              (1,"b","8" ),
              (2, "c", "3"),
              (3,"r", "19")).toDF("col1", "col2", "col3"))

df.show()
+----+----+----+
|col1|col2|col3|
+----+----+----+
|   1|   a|   1|
|   1|   b|   3|
|   1|   c|   6|
|   2|   a|   9|
|   2|   c|  10|
|   1|   b|   8|
|   2|   c|   3|
|   3|   r|  19|
+----+----+----+

我需要按 col1 和 col2 分组并计算 col3 的平均值，我可以使用：

val col1df = df.groupBy("col1").agg(round(mean("col3"),2).alias("mean_col1"))
val col2df = df.groupBy("col2").agg(round(mean("col3"),2).alias("mean_col2"))

但是，在具有几百万行和几万列中要分组的唯一元素的大型数据帧上，这需要很长时间。此外，我还有更多列要分组，而且花费的时间非常长，我希望减少这些列。有没有更好的方法来做 groupBy 后跟聚合？

【问题讨论】：

标签： scala apache-spark group-by aggregate aggregate-functions

【解决方案1】：

您可以使用Multiple Aggregations 的想法，它可能会在一次 shuffle 操作中完成所有操作，这是最昂贵的操作。

例子：

val df = (Seq((1, "a", "1"),
(1,"b", "3"),
(1,"c", "6"),
(2, "a", "9"),
(2,"c", "10"),
(1,"b","8" ),
(2, "c", "3"),
(3,"r", "19")).toDF("col1", "col2", "col3"))

df.createOrReplaceTempView("data")

val grpRes = spark.sql("""select grouping_id() as gid, col1, col2, round(mean(col3), 2) as res 
                          from data group by col1, col2 grouping sets ((col1), (col2)) """)

grpRes.show(100, false)

输出：

+---+----+----+----+
|gid|col1|col2|res |
+---+----+----+----+
|1  |3   |null|19.0|
|2  |null|b   |5.5 |
|2  |null|c   |6.33|
|1  |1   |null|4.5 |
|2  |null|a   |5.0 |
|1  |2   |null|7.33|
|2  |null|r   |19.0|
+---+----+----+----+

gid 使用起来有点有趣，因为它下面有一些二进制计算。但是，如果您的分组列不能有空值，那么您可以使用它来选择正确的组。

执行计划：

scala> grpRes.explain
== Physical Plan ==
*(2) HashAggregate(keys=[col1#111, col2#112, spark_grouping_id#108], functions=[avg(cast(col3#9 as double))])
+- Exchange hashpartitioning(col1#111, col2#112, spark_grouping_id#108, 200)
   +- *(1) HashAggregate(keys=[col1#111, col2#112, spark_grouping_id#108], functions=[partial_avg(cast(col3#9 as double))])
      +- *(1) Expand [List(col3#9, col1#109, null, 1), List(col3#9, null, col2#110, 2)], [col3#9, col1#111, col2#112, spark_grouping_id#108]
         +- LocalTableScan [col3#9, col1#109, col2#110]

如您所见，只有单一的 Exchange 操作，代价高昂的 shuffle。

【讨论】：