Spark多个动态聚合函数，countDistinct不起作用答案

【问题标题】：Spark multiple dynamic aggregate functions, countDistinct not workingSpark多个动态聚合函数，countDistinct不起作用
【发布时间】：2019-09-02 13:45:06
【问题描述】：

具有多个动态聚合操作的 Spark 数据帧上的聚合。

我想使用具有多个动态聚合操作（由用户以 JSON 格式传递）的 Scala 对 Spark 数据帧进行聚合。我正在将 JSON 转换为 Map。

以下是一些示例数据：

colA    colB    colC    colD
1       2       3       4
5       6       7       8
9       10      11      12

我正在使用的 Spark 聚合代码：

var cols = ["colA","colB"]
var aggFuncMap = Map("colC"-> "sum", "colD"-> "countDistinct")
var aggregatedDF = currentDF.groupBy(cols.head, cols.tail: _*).agg(aggFuncMap)

我必须将aggFuncMap 仅作为Map 传递，以便用户可以通过 JSON 配置传递任意数量的聚合。

上述代码适用于某些聚合，包括sum、min、max、avg 和count。

但是，不幸的是，此代码不适用于countDistinct（可能是因为它是驼峰式的？）。

运行上述代码时，出现以下错误：

线程“主”org.apache.spark.sql.AnalysisException 中的异常：未定义函数：'countdistinct'。该函数既不是注册的临时函数，也不是在数据库'default'中注册的永久函数

任何帮助将不胜感激！

【问题讨论】：

标签： scala apache-spark count apache-spark-sql distinct

【解决方案1】：

目前无法在Map 内使用agg 和countDistinct。从documentation我们看到：

可用的聚合方法有 avg、max、min、sum、count。

一种可能的解决方法是将Map 更改为Seq[Column]，

val cols = Seq("colA", "colB")
val aggFuncs = Seq(sum("colC"), countDistinct("colD"))
val df2 = df.groupBy(cols.head, cols.tail: _*).agg(aggFuncs.head, aggFuncs.tail: _*)

但如果用户要在配置文件中指定聚合，那将无济于事。

另一种方法是使用expr，该函数将评估一个字符串并返回一个列。但是，expr 不会接受"countDistinct"，而是需要使用"count(distinct(...))"。可以这样编码：

val aggFuncs = Seq("sum(colC)", "count(distinct(colD))").map(e => expr(e))
val df2 = df.groupBy(cols.head, cols.tail: _*).agg(aggFuncs.head, aggFuncs.tail: _*)

【讨论】：