Spark 2.3.0(Scala2.11)自定义Aggregator：如何为Scala集合创建Encoder？

我在Spark 2.x版本开发自定义Aggregator时也碰到过一模一样的问题，当时缓冲区用了可变集合，不知道怎么给它生成Encoder。给你两个实用的解决方案，都是基于Spark 2.3.0 + Scala 2.11环境验证过的：

方案一：利用List的Encoder转换（最简洁） #

Spark本身没有直接提供ListBuffer的Encoder，但它对List有内置的支持。我们可以通过Encoder.map()方法，把List的Encoder转换成ListBuffer的Encoder——这个方法允许我们定义编码（从ListBuffer转List）和解码（从List转ListBuffer）的逻辑：

import org.apache.spark.sql.Encoder
import scala.collection.mutable.ListBuffer

case class Foo(/* 你的字段定义 */)

object MyAggregator extends Aggregator[Foo, ListBuffer[Foo], Boolean] {
  // 确保已经导入SparkSession的implicits，比如在调用Aggregator的地方：import spark.implicits._
  
  override def bufferEncoder: Encoder[ListBuffer[Foo]] = {
    // 先获取List[Foo]的隐式Encoder
    val listEncoder = implicitly[Encoder[List[Foo]]]
    // 转换为ListBuffer[Foo]的Encoder：编码时转List，解码时转ListBuffer
    listEncoder.map(_.toListBuffer, _.toList)
  }

  // 其他必须重写的方法：zero, reduce, merge, finish
  override def zero: ListBuffer[Foo] = ListBuffer.empty[Foo]
  
  override def reduce(buffer: ListBuffer[Foo], input: Foo): ListBuffer[Foo] = {
    buffer += input
    buffer
  }
  
  override def merge(b1: ListBuffer[Foo], b2: ListBuffer[Foo]): ListBuffer[Foo] = {
    b1 ++= b2
    b1
  }
  
  override def finish(buffer: ListBuffer[Foo]): Boolean = {
    // 这里写你的最终计算逻辑，比如判断历史行是否满足某个条件
    buffer.nonEmpty // 示例逻辑，替换成你的需求
  }
  
  override def outputEncoder: Encoder[Boolean] = implicitly[Encoder[Boolean]]
}

方案二：手动创建ExpressionEncoder（更灵活） #

如果需要更精细的控制，也可以用ExpressionEncoder手动构建。因为Foo是case class，Spark能自动生成它的Encoder，我们可以基于这个来构建ListBuffer的Encoder：

import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
import org.apache.spark.sql.types.ArrayType
import scala.collection.mutable.ListBuffer

override def bufferEncoder: Encoder[ListBuffer[Foo]] = {
  val fooEncoder = implicitly[Encoder[Foo]]
  val arrayType = ArrayType(fooEncoder.schema)
  
  ExpressionEncoder[ListBuffer[Foo]](
    schema = arrayType,
    flat = false,
    serializer = { obj =>
      // 把ListBuffer转成Array（或者List）来序列化
      StaticInvoke(
        classOf[ListBuffer[_]],
        fooEncoder.schema,
        "toArray",
        obj :: Nil
      )
    },
    deserializer = { row =>
      // 把Array转成ListBuffer
      StaticInvoke(
        classOf[ListBuffer[_]],
        ExpressionEncoder.boxedType[ListBuffer[Foo]],
        "apply",
        row :: Nil
      ).asInstanceOf[ListBuffer[Foo]]
    }
  )
}

不过这个方法代码量更大，一般方案一就足够满足需求了。

注意事项 #

• 一定要确保代码中导入了spark.implicits._（spark是你的SparkSession实例），否则implicitly[Encoder[List[Foo]]]会找不到对应的隐式值。
• 因为ListBuffer是可变集合，在merge方法里要正确合并两个缓冲区（比如用++=），避免出现数据丢失或者并发问题——Spark在执行Aggregator时，每个分区内的缓冲区是单线程处理的，所以只要merge逻辑正确就不会有线程安全问题。

内容的提问来源于stack exchange，提问作者Uncle Long Hair