Scala中的Spark SQL(V2.0)UDAF返回空字符串

Question

当我试图为我们的一个复杂问题创建一个UDAF时，我决定从一个基本的UDAF开始，该UDAF按原样返回列。既然我是刚开始使用SQL/Scala的，有人能帮我指出我的错误吗。

代码如下：

导入org.apache.spark.sql.expressions.mutableaggregationbuffer导入org.apache.spark.sql.expressions.userdefinedaggregatefunction导入org.apache.spark.sql.row导入org.apache.spark.sql.types._导入org.apache.spark.sql.types.dataTypes

导入Scala.Collection._

重写def bufferSchema：StructType=StructType（StructField（“shingles”,（StringType））：：Nil）

重写def DataType：DataType=（StringType）

重写def deterministic：Boolean=true

重写def merge(buffer1:MutableAggregationBuffer,buffer2:Row):Unit={}

重写def evaluate（buffer：Row）：Any={buffer（0）}}

运行上述UDAF的代码如下：

import spark.implicits._;

val df = spark.read.json("people.json")
df.createOrReplaceTempView("people")
val sqlDF = spark.sql("Select name from people")
sqlDF.show()

val minhash = df.select(MinhashUdaf(col("name")).as("minhash"))
minhash.printSchema()
minhash.show(truncate = false)

由于在UDAF中我是按原样返回输入，所以我应该按原样获取每一行的列“name”的值。而在运行上面的字符串时,返回的是一个空字符串。

Answer 1

您没有实现合并函数。

使用下面的代码，可以根据需要打印列的值。

object MinhashUdaf extends UserDefinedAggregateFunction {

override def inputSchema: org.apache.spark.sql.types.StructType = StructType(StructField("value", StringType) :: Nil)

override def bufferSchema: StructType = StructType( StructField("shingles", (StringType)) :: Nil)

override def dataType: DataType = (StringType)

override def deterministic: Boolean = true

override def initialize(buffer: MutableAggregationBuffer): Unit = { buffer(0) = ("") }

override def update(buffer: MutableAggregationBuffer, input: Row): Unit = { buffer.update(0, input.get(0)) }

override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {  buffer1.update(0, buffer2.get(0))}

override def evaluate(buffer: Row): Any = { buffer(0) } }