用户自定义函数UDAF_大数据培训

UDAF

强类型的Dataset和弱类型的DataFrame都提供了相关的聚合函数， 如 count()，countDistinct()，avg()，max()，min()。除此之外，用户可以设定自己的自定义聚合函数。通过继承UserDefinedAggregateFunction来实现用户自定义聚合函数。

1）需求：实现求平均工资的自定义聚合函数。

2）代码实现

import org.apache.spark.sql.expressions.MutableAggregationBuffer

import org.apache.spark.sql.expressions.UserDefinedAggregateFunction

import org.apache.spark.sql.types._

import org.apache.spark.sql.Row

import org.apache.spark.sql.SparkSession

object MyAverage extends UserDefinedAggregateFunction {

// 聚合函数输入参数的数据类型

def inputSchema: StructType = StructType(StructField(“inputColumn”, LongType) :: Nil)

// 聚合缓冲区中值得数据类型

def bufferSchema: StructType = {

StructType(StructField(“sum”, LongType) :: StructField(“count”, LongType) :: Nil)

}

// 返回值的数据类型

def dataType: DataType = DoubleType

// 对于相同的输入是否一直返回相同的输出。

def deterministic: Boolean = true

// 初始化

def initialize(buffer: MutableAggregationBuffer): Unit = {

// 存工资的总额

buffer(0) = 0L

// 存工资的个数

buffer(1) = 0L

}

// 同一个分区数据合并。

def update(buffer: MutableAggregationBuffer, input: Row): Unit = {

if (!input.isNullAt(0)) {

buffer(0) = buffer.getLong(0) + input.getLong(0)

buffer(1) = buffer.getLong(1) + 1

}

}

// 不同分区间数据合并

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

buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)

buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)

}

// 计算最终结果

def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1)

}

3）函数使用