Spark Scala中Dataset调用agg函数提示无法解析重载方法问题

你遇到的报错是因为groupByKey返回的是**KeyValueGroupedDataset**，它的agg方法和DataFrame的agg参数要求完全不同——不能直接传入org.apache.spark.sql.functions.avg这类面向DataFrame的函数，这也是为什么你参考的DataFrame教程写法在这里不生效。

下面给你两种可行的解决思路：

方法一：转成DataFrame处理（简单直接） #

把map后的Dataset转换成DataFrame，就能用你熟悉的SQL聚合函数了：

import org.apache.spark.sql.functions.avg

val calculatedAverage = dataset
  .map(mapFunctionToTheSchema)
  .toDF("key", "value") // 指定列名，也可以让Spark自动推断
  .groupBy("key")
  .agg(avg("value").as("average"))

// 如果需要转回强类型Dataset，可追加：
// .as[(Long, Double)]

方法二：使用Dataset原生的Typed聚合（类型安全） #

如果想保持Dataset的强类型特性，可以自定义一个Typed聚合器：

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

// 定义聚合状态的样例类
case class AvgState(sum: Long, count: Long)

// 实现计算平均值的Typed聚合器
object Average extends Aggregator[(Long, Long), AvgState, Double] {
  // 初始化聚合状态
  override def zero: AvgState = AvgState(0L, 0L)
  // 累加单个元素到状态中
  override def reduce(state: AvgState, value: (Long, Long)): AvgState =
    AvgState(state.sum + value._2, state.count + 1)
  // 合并不同分区的聚合状态
  override def merge(a: AvgState, b: AvgState): AvgState =
    AvgState(a.sum + b.sum, a.count + b.count)
  // 计算最终平均值
  override def finish(reduction: AvgState): Double =
    if (reduction.count == 0) 0.0 else reduction.sum.toDouble / reduction.count
  // 状态的编码器
  override def bufferEncoder: Encoder[AvgState] = Encoders.product
  // 输出结果的编码器
  override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

// 使用自定义聚合器完成分组计算
val calculatedAverage = dataset
  .map(mapFunctionToTheSchema)
  .groupByKey(_.key)
  .agg(Average.toColumn.name("average"))

内容的提问来源于stack exchange，提问作者Hubert Bratek