Spark 3.0.1是否支持窗口函数上的自定义聚合器？

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

case class Data(value: Int)

case class Result(sum: Int, count: Int)

class CustomAggregator extends Aggregator[Data, Result, Double] {
  override def zero: Result = Result(0, 0)

  override def reduce(b: Result, a: Data): Result = {
    Result(b.sum + a.value, b.count + 1)
  }

  override def merge(b1: Result, b2: Result): Result = {
    Result(b1.sum + b2.sum, b1.count + b2.count)
  }

  override def finish(reduction: Result): Double = {
    reduction.sum.toDouble / reduction.count
  }

  override def bufferEncoder: Encoder[Result] = Encoders.product

  override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

object WindowFunctionExample {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("WindowFunctionExample")
      .master("local")
      .getOrCreate()

    import spark.implicits._

    val data = Seq(Data(1), Data(2), Data(3), Data(4), Data(5))
    val dataset = spark.createDataset(data)

    val customAggregator = new CustomAggregator()

    val result = dataset.selectExpr("value")
      .groupBy($"value")
      .agg(customAggregator.toColumn)
      .show()

    spark.stop()
  }
}