spark udaf 更新数组类型的元组

Question

我正在使用 Scala + Spark 2.0 并尝试编写一个 UDAF，其中包含一个元组数组作为其内部缓冲区以及它的返回类型： ...

def bufferSchema = new StructType().add("midResults", ArrayType(  StructType(Array(StructField("a", DoubleType),StructField("b", DoubleType))) ))

def dataType: DataType = ArrayType(  StructType(Array(StructField( "a", DoubleType),StructField("b", DoubleType))) )

这就是我更新缓冲区的方式

def update(buffer: MutableAggregationBuffer, input: Row) = {
buffer(0) = buffer.getAs[mutable.WrappedArray[(Double,Double)]](3) ++ Array((3.0,4.0))
}

但我得到以下异常：

java.lang.ArrayStoreException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema

如果我有一个简单的 Double 数组，则此模式有效..

Answer 1

java.lang.ArrayStoreException 是 "thrown to indicate that an attempt has been made to store the wrong type of object into an array of objects" 并且预期的 because a local Scala type for StructType is o.a.s.sql.Row 不是元组。换句话说，您应该使用Seq[Row] 作为缓冲区字段并使用Row 作为值。

注意事项：

• 循环调用++ 可能不是最好的主意。
• 如果您认为创建 UDAF 有点过时，因为 Spark 2.0 collect_list 支持复杂类型。
• 可以说Aggregators 比UserDefinedAggregateFunctions 更加用户友好。