Scala Spark自定义Aggregator聚合多列时出现id列解析异常

核心原因 #

你遇到的错误本质是Spark聚合阶段的列作用域限制，以及自定义Aggregator输入处理逻辑的错误：

1. 当使用struct(id, hash)作为聚合输入时，原id、hash列会被封装成一个单一的Struct对象，不再是聚合上下文里的顶级列。如果你的Aggregator代码直接引用id列名（比如排序逻辑中），就会因超出作用域导致解析失败。
2. 改用Row作为输入时，若未正确匹配Row的字段顺序、类型，或未通过索引/字段名从Row中提取值，同样会引发解析错误。
3. Spark聚合阶段只能访问你传递给Aggregator的输入参数，无法直接引用原DataFrame的列名，所有需要的数据必须从输入参数中提取。

正确实现示例 #

以下是适配Spark 3.3.x+的自定义Aggregator实现，按id排序后拼接hash并生成总哈希值：

方式1：用Case Class作为输入类型（可读性更强）

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

// 定义输入数据结构
case class IdHashPair(id: Long, hash: String)

class SortedHashAggregator extends Aggregator[IdHashPair, List[(Long, String)], String] {
  // 初始化空缓冲区
  override def zero: List[(Long, String)] = List.empty

  // 单条数据合并到缓冲区
  override def reduce(acc: List[(Long, String)], input: IdHashPair): List[(Long, String)] = {
    acc :+ (input.id, input.hash)
  }

  // 合并两个缓冲区
  override def merge(acc1: List[(Long, String)], acc2: List[(Long, String)]): List[(Long, String)] = {
    acc1 ++ acc2
  }

  // 生成最终结果：按id排序后拼接hash，再计算总哈希
  override def finish(reduction: List[(Long, String)]): String = {
    val sortedHashes = reduction.sortBy(_._1).map(_._2).mkString("")
    // 这里用MD5示例，可替换为你需要的哈希算法
    java.security.MessageDigest.getInstance("MD5")
      .digest(sortedHashes.getBytes)
      .map("%02x".format(_))
      .mkString
  }

  // 编码器定义
  override def bufferEncoder: Encoder[List[(Long, String)]] = Encoders.kryo[List[(Long, String)]]
  override def outputEncoder: Encoder[String] = Encoders.STRING
  override def inputEncoder: Encoder[IdHashPair] = Encoders.product[IdHashPair]
}

调用方式

import org.apache.spark.sql.functions._

// 示例DataFrame
val df = spark.createDataFrame(Seq(
  (1L, "hash1"),
  (2L, "hash2"),
  (1L, "hash3")
)).toDF("id", "hash")

// 实例化聚合器并调用
val sortedHashAgg = new SortedHashAggregator().toColumn.name("sorted_combined_hash")

// 按需求分组（示例为全局聚合，可替换为实际分组列）
df.groupBy(lit(1))
  .agg(sortedHashAgg.of(struct(col("id"), col("hash"))))
  .show()

方式2：用Tuple作为输入类型（更简洁）

如果不想定义Case Class，可直接用Tuple作为输入：

class SortedHashTupleAggregator extends Aggregator[(Long, String), List[(Long, String)], String] {
  override def zero: List[(Long, String)] = List.empty
  override def reduce(acc: List[(Long, String)], input: (Long, String)): List[(Long, String)] = acc :+ input
  override def merge(acc1: List[(Long, String)], acc2: List[(Long, String)]): List[(Long, String)] = acc1 ++ acc2
  override def finish(reduction: List[(Long, String)]): String = {
    val sortedHashes = reduction.sortBy(_._1).map(_._2).mkString("")
    java.security.MessageDigest.getInstance("MD5")
      .digest(sortedHashes.getBytes)
      .map("%02x".format(_))
      .mkString
  }
  override def bufferEncoder: Encoder[List[(Long, String)]] = Encoders.kryo[List[(Long, String)]]
  override def outputEncoder: Encoder[String] = Encoders.STRING
  override def inputEncoder: Encoder[(Long, String)] = Encoders.tuple(Encoders.LONG, Encoders.STRING)
}

// 调用时直接传入列
val sortedHashTupleAgg = new SortedHashTupleAggregator().toColumn.name("sorted_combined_hash")
df.groupBy(lit(1))
  .agg(sortedHashTupleAgg.of(col("id"), col("hash")))
  .show()

关键注意点 #

• 所有需要用到的id、hash数据，必须从Aggregator的输入参数（Case Class/Tuple/Row）中提取，不能直接引用原DataFrame的列名。
• 若使用Row作为输入，需严格匹配字段顺序和类型，比如通过input.getLong(0)获取id，input.getString(1)获取hash，同时确保输入Encoder正确配置。

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