Spark 3.0 已弃用UserDefinedAggregateFunction
,我试图使用Aggregator
. 的基本用法Aggregator
很简单,但是,我很难使用更通用的函数版本。
我将尝试用这个例子来解释我的问题,一个collect_set
. 这不是我的实际情况,但更容易解释问题:
class CollectSetDemoAgg(name: String) extends Aggregator[Row, Set[Int], Set[Int]] {
override def zero = Set.empty
override def reduce(b: Set[Int], a: Row) = b + a.getInt(a.fieldIndex(name))
override def merge(b1: Set[Int], b2: Set[Int]) = b1 ++ b2
override def finish(reduction: Set[Int]) = reduction
override def bufferEncoder = Encoders.kryo[Set[Int]]
override def outputEncoder = ExpressionEncoder()
}
// using it:
df.agg(new CollectSetDemoAgg("rank").toColumn as "result").show()
我更喜欢.toColumn
vs .udf.register
,但这不是重点。
问题: 我不能制作这个聚合器的通用版本,它只适用于整数。
我尝试过:
class CollectSetDemo(name: String) extends Aggregator[Row, Set[Any], Set[Any]]
它因错误而崩溃:
No Encoder found for Any
- array element class: "java.lang.Object"
- root class: "scala.collection.immutable.Set"
java.lang.UnsupportedOperationException: No Encoder found for Any
- array element class: "java.lang.Object"
- root class: "scala.collection.immutable.Set"
at org.apache.spark.sql.catalyst.ScalaReflection$.$anonfun$serializerFor$1(ScalaReflection.scala:567)
我不能去CollectSetDemo[T]
,万一我不能正常outputEncoder
。此外,在使用 udaf 时,我只能使用 Spark 数据类型、列等。