如何将数组传递给Spark（UDAF）中的用户定义聚合函数

Question

我想在UDAF中传递一个数组作为输入模式。

我给出的例子非常简单，它只是对2个向量求和。实际上我的用例更复杂，我需要使用UDAF。

import sc.implicits._
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import org.apache.spark.sql.expressions._

val df = Seq(
  (1, Array(10.2, 12.3, 11.2)),
  (1, Array(11.2, 12.6, 10.8)),
  (2, Array(12.1, 11.2, 10.1)),
  (2, Array(10.1, 16.0, 9.3)) 
  ).toDF("siteId", "bidRevenue")


class BidAggregatorBySiteId() extends UserDefinedAggregateFunction {

  def inputSchema: StructType = StructType(Array(StructField("bidRevenue", ArrayType(DoubleType))))

  def bufferSchema = StructType(Array(StructField("sumArray", ArrayType(DoubleType))))

  def dataType: DataType = ArrayType(DoubleType)

  def deterministic = true

  def initialize(buffer: MutableAggregationBuffer) = {
      buffer.update(0, Array(0.0, 0.0, 0.0))
      }

  def update(buffer: MutableAggregationBuffer, input: Row) = {
      val seqBuffer = buffer(0).asInstanceOf[IndexedSeq[Double]]
      val seqInput = input(0).asInstanceOf[IndexedSeq[Double]]
      buffer(0) = seqBuffer.zip(seqInput).map{ case (x, y) => x + y }
  }

  def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
     val seqBuffer1 = buffer1(0).asInstanceOf[IndexedSeq[Double]]
     val seqBuffer2 = buffer2(0).asInstanceOf[IndexedSeq[Double]]
     buffer1(0) = seqBuffer1.zip(seqBuffer2).map{ case (x, y) => x + y }
  }

  def evaluate(buffer: Row) = { 
    buffer
  }
}
val fun = new BidAggregatorBySiteId()

df.select($"siteId", $"bidRevenue" cast(ArrayType(DoubleType)))
.groupBy("siteId").agg(fun($"bidRevenue"))
.show

在“显示”动作之前，所有这些都可以很好地进行转换。但这部剧引发了一个错误：

斯卡拉。MatchError:[WrappedArray（21.4，24.9，22.0）]（属于org.apache.spark.sql.execution.aggregate.InputAggregationBuffer类）。阿帕奇。火花sql。催化剂CatalystTypeConverters$ArrayConverter。toCatalystImpl（CatalystTypeConverters.scala:160）

我的数据帧的结构是：

root
 |-- siteId: integer (nullable = false)
 |-- bidRevenue: array (nullable = true)
 |    |-- element: double (containsNull = true)

df。dtypes=数组[（String，String）]=数组（（“siteId”，“IntegerType”），（“bidRevenue”，“ArrayType（DoubleType，true）”）

坦克为你提供了宝贵的帮助。

Answer 1

def evaluate(buffer: Row): Any

一旦一个组被完全处理以获得最终结果，就会调用上述方法。当您仅初始化和更新缓冲区的第0个索引时

i.e. buffer(0)  

因此，您需要在最后返回第0个索引值，因为您的聚合结果存储在0索引处。

  def evaluate(buffer: Row) = {
    buffer.get(0)
  }

上述对评估（）方法的修改将导致：

// +------+---------------------------------+
// |siteId|bidaggregatorbysiteid(bidRevenue)|
// +------+---------------------------------+
// |     1|               [21.4, 24.9, 22.0]|
// |     2|               [22.2, 27.2, 19.4]|
// +------+---------------------------------+