object UDAFApp extends App{val spark = SparkSession.builder().master("local[2]").appName("UDAP-App").getOrCreate();import  spark.implicits._;val df = spark.read.format("json").load("D:\\data\\employees.json")//UDAF函数注册 只有UserDefinedAggregateFunction才能为SQL注册函数spark.udf.register("cusAvg",MyAvgUDAF)//DF转临时视图df.createTempView("employees_view")spark.sql("select cusAvg(salary) as salary from employees_view").show();//df-api形式df.select(MyAvgUDAF.apply($"salary")).show()spark.close()}object MyAvgUDAF extends UserDefinedAggregateFunction{//输入schemaoverride def inputSchema: StructType = StructType(StructField("input",DoubleType)::Nil);//聚合过程schemaoverride def bufferSchema: StructType = StructType(StructField("Sum",DoubleType)::StructField("Count",LongType)::Nil)//返回值类型override def dataType: DataType = DoubleType//是否固定返回值类型override def deterministic: Boolean = true//初始化函数override def initialize(buffer: MutableAggregationBuffer): Unit = {//设定聚合基准初始值 aggregate算子((0,0))的部分buffer(0) = 0D; //总和0buffer(1) = 0L; //个数0
            }override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {//行第一列(Row[0])是否为nullif(!input.isNullAt(0)){//aggregate算子....(seqOp: (U, T) => U 部分buffer(0)= buffer.getDouble(0)+ input.getDouble(0);buffer(1) =buffer.getLong(1)+1;}}override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {//aggregate算子....combOp: (U, U) => U 部分buffer1(0) = buffer1.getDouble(0) + buffer2.getDouble(0);buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1);}override def evaluate(buffer: Row): Any = buffer.getDouble(0) /buffer.getLong(1) ;}

            object UDAFApp extends App{val spark = SparkSession.builder().master("local[2]").appName("UDAP-App").getOrCreate();import  spark.implicits._;val ds = spark.read.format("json").load("D:\\data\\employees.json").as[Employee]//ds-api形式ds.select(MyAverage.toColumn.name("salary")).show()spark.close()}//目标类型定义case class Employee(val name: String,val salary: Long)//聚合类型定义case class Average(var sum: Long, var count: Long)object MyAverage extends Aggregator[Employee, Average, Double]  {override def zero: Average = Average(0,0)override def reduce(b: Average, a: Employee): Average = {b.sum += a.salary;b.count +=  1b}override def merge(b1: Average, b2: Average): Average = {b1.sum += b2.sum;b1.count += b2.count;b1;}override def finish(reduction: Average): Double = {println(reduction.sum + "  "+ reduction.count)reduction.sum.toDouble/reduction.count}override def bufferEncoder: Encoder[Average] = Encoders.productoverride def outputEncoder: Encoder[Double] = Encoders.scalaDouble}