感謝我的同事 李震給我講解UDAF

網(wǎng)上找到的大部分都只有代碼，但是缺少講解，官網(wǎng)的的API有講解，但是看不太明白。我還是自己記錄一下吧，或許對(duì)其他人有幫助。

接下來(lái)以一個(gè)求幾何平均數(shù)的例子來(lái)說(shuō)明如何實(shí)現(xiàn)一個(gè)自己的UDAF

?

首先需要導(dǎo)入這些包：

import org.apache.spark.sql.expressions.MutableAggregationBuffer import org.apache.spark.sql.expressions.UserDefinedAggregateFunction import org.apache.spark.sql.Row import org.apache.spark.sql.types._
需要繼承實(shí)現(xiàn)這個(gè)抽象類 class GeometricMean extends UserDefinedAggregateFunction { // This is the input fields for your aggregate function.
就是需要輸入的列的類型，可以有多個(gè)列，多個(gè)列的寫(xiě)法如下：
/*
StructType(StructField("slot",IntegerType) :: StructField("score",IntegerType)::Nil) */ override def inputSchema: org.apache.spark.sql.types.StructType = StructType(StructField("value", DoubleType) :: Nil)

存儲(chǔ)聚合結(jié)果的中間buffer // This is the internal fields you keep for computing your aggregate. override def bufferSchema: StructType = StructType( StructField("count", LongType) :: StructField("product", DoubleType) :: Nil ) // This is the output type of your aggregatation function.
返回結(jié)果的類型，比如這個(gè)集合平均數(shù)就是返回一個(gè)double值 override def dataType: DataType = DoubleType
是每次運(yùn)行結(jié)果都過(guò)一樣，但是我也不太明白啊 override def deterministic: Boolean = true

初始化存儲(chǔ)聚合結(jié)果的buffer // This is the initial value for your buffer schema. override def initialize(buffer: MutableAggregationBuffer): Unit = { buffer(0) = 0L buffer(1) = 1.0 }
每次更新怎么更新，比如新來(lái)了一行，如何加入更新聚合的結(jié)果 // This is how to update your buffer schema given an input. override def update(buffer: MutableAggregationBuffer, input: Row): Unit = { buffer(0) = buffer.getAs[Long](0) + 1 buffer(1) = buffer.getAs[Double](1) * input.getAs[Double](0) }
spark會(huì)把數(shù)據(jù)劃分成多個(gè)塊，每個(gè)塊都會(huì)進(jìn)行處理，然后把每個(gè)塊的結(jié)果進(jìn)行合并處理 // This is how to merge two objects with the bufferSchema type. override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = { buffer1(

轉(zhuǎn)載于:https://www.cnblogs.com/earendil/p/8510680.html

總結(jié)

以上是生活随笔為你收集整理的spark UDAF的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。