Spark 数据帧的自定义聚合答案

【问题标题】：Custom aggregations for Spark dataframesSpark 数据帧的自定义聚合
【发布时间】：2016-10-09 11:53:29
【问题描述】：

我想知道是否有某种方法可以为 Spark 数据帧指定自定义聚合函数。如果我有一个包含 2 列 id 和 value 的表，我想 groupBy id 并将值聚合到每个 value 的列表中，如下所示：

来自：

john | tomato
john | carrot
bill | apple
john | banana
bill | taco

到：

john | tomato, carrot, banana
bill | apple, taco

这在数据帧中可行吗？我在询问数据框，因为我正在将数据作为 orc 文件读取，并且它作为数据框加载。我认为将其转换为 RDD 效率低下。

【问题讨论】：

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

【解决方案1】：

我只是简单地使用以下内容：

import org.apache.spark.sql.functions.collect_list
val df = Seq(("john", "tomato"), ("john", "carrot"), 
             ("bill", "apple"), ("john", "banana"), 
             ("bill", "taco")).toDF("id", "value")
// df: org.apache.spark.sql.DataFrame = [id: string, value: string]

val aggDf = df.groupBy($"id").agg(collect_list($"value").as("values"))
// aggDf: org.apache.spark.sql.DataFrame = [id: string, values: array<string>]

aggDf.show(false)
// +----+------------------------+
// |id  |values                  |
// +----+------------------------+
// |john|[tomato, carrot, banana]|
// |bill|[apple, taco]           |
// +----+------------------------+

您甚至不需要调用底层的rdd。

【讨论】：

【解决方案2】：

恢复到RDD 操作往往最适合解决此类问题：

scala> val df = sc.parallelize(Seq(("john", "tomato"),
           ("john", "carrot"), ("bill", "apple"), 
           ("john", "bannana"), ("bill", "taco")))
           .toDF("name", "food")
df: org.apache.spark.sql.DataFrame = [name: string, food: string]

scala> df.show
+----+-------+
|name|   food|
+----+-------+
|john| tomato|
|john| carrot|
|bill|  apple|
|john|bannana|
|bill|   taco|
+----+-------+

scala> val aggregated = df.rdd
           .map{ case Row(k: String, v: String) => (k, List(v)) }
           .reduceByKey{_ ++ _}
           .toDF("name", "foods")
aggregated: org.apache.spark.sql.DataFrame = [name: string, foods: array<string>]

scala> aggregated.collect.foreach{println}
[john,WrappedArray(tomato, carrot, bannana)]
[bill,WrappedArray(apple, taco)]

至于效率，我相信 DataFrames 是 RDDs 在后台，所以像 .rdd 这样的转换成本非常低。

【讨论】：