在多个列上应用自定义 Spark 聚合器 (Spark 2.0)答案

【问题标题】：Apply a custom Spark Aggregator on multiple columns (Spark 2.0)在多个列上应用自定义 Spark 聚合器 (Spark 2.0)
【发布时间】：2017-05-19 22:07:35
【问题描述】：

我为字符串创建了一个自定义Aggregator[]。

我想将它应用于DataFrame 的所有列，其中所有列都是字符串，但列号是任意的。

我一直在写正确的表达方式。我想写这样的东西：

df.agg( df.columns.map( c => myagg(df(c)) ) : _*)

考虑到各种接口，这显然是错误的。

我查看了RelationalGroupedDataset.agg(expr: Column, exprs: Column*) 代码，但我不熟悉表达式操作。

有什么想法吗？

【问题讨论】：

请出示您的聚合代码。并解释你想要做什么。
@AssafMendelson，实际上我们计划为各种数据类型的各种统计数据创建许多自定义聚合器。我从一个聚合器开始，以获得最短和最长的字符串： class ShortestLongestAggregator() extends Aggregator[String, (String, String), (String, String)] 。现在我想要一个任意数据框的所有列的所有（最短，最长）对（假设它只有字符串列）。

标签： apache-spark apache-spark-sql aggregate-functions user-defined-functions

【解决方案1】：

与UserDefinedAggregateFunctions 对单个字段（列）进行操作相比，Aggregtors 需要一个完整的Row / 值。

如果你想要 Aggregator 可以在你的 sn-p 中使用，它必须由列名参数化并使用 Row 作为值类型。

import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{Encoder, Encoders, Row}

case class Max(col: String) 
    extends Aggregator[Row, Int, Int] with Serializable {

  def zero = Int.MinValue
  def reduce(acc: Int, x: Row) =
    Math.max(acc, Option(x.getAs[Int](col)).getOrElse(zero))

  def merge(acc1: Int, acc2: Int) = Math.max(acc1, acc2)
  def finish(acc: Int) = acc

  def bufferEncoder: Encoder[Int] = Encoders.scalaInt
  def outputEncoder: Encoder[Int] = Encoders.scalaInt
}

示例用法：

val df = Seq((1, None, 3), (4, Some(5), -6)).toDF("x", "y", "z")

@transient val exprs = df.columns.map(c => Max(c).toColumn.alias(s"max($c)"))

df.agg(exprs.head, exprs.tail: _*)

+------+------+------+
|max(x)|max(y)|max(z)|
+------+------+------+
|     4|     5|     3|
+------+------+------+

可以说Aggregators 与静态类型的Datasets 结合使用比Dataset<Row> 更有意义。

根据您的要求，您还可以使用Seq[_] 累加器一次聚合多个列，并在单个merge 调用中处理整个Row（记录）。

【讨论】：