1）创建弱类型聚合函数类extends UserDefinedAggregateFunction

class MyAgeFunction extends UserDefinedAggregateFunction {//函数输入的数据结构，需要new一个具体的结构对象，然后添加结构override def inputSchema: StructType = {new StructType().add("age",LongType)}//计算时的数据结构override def bufferSchema: StructType = {new StructType().add("sum",LongType).add("conut",LongType)}//函数返回的数据类型override def dataType: DataType = DoubleType//表述函数是否稳定override def deterministic: Boolean = true//表述的是函数计算之前的缓冲区的初始化 buffer(0)表示第一个结构：sum， buffer(1)示第二个结构：countoverride def initialize(buffer: MutableAggregationBuffer): Unit = {buffer(0) = 0Lbuffer(1) = 0L}//根据查询结构来更新缓冲区数据sum + = input.getLong  count+=1override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {buffer(0) = buffer.getLong(0) + input.getLong(0)buffer(1) = buffer.getLong(1) + 1}//将多个节点的缓冲区合并override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)}//计算override def evaluate(buffer: Row): Any = {buffer.getLong(0).toDouble / buffer.getLong(1)}
}

聚合函数使用

def main(args: Array[String]): Unit = {//创建配置对象val conf = new SparkConf().setAppName("Spark01_Custom").setMaster("local[*]")val spark = SparkSession.builder().config(conf).getOrCreate()val rdd1 = spark.sparkContext.makeRDD(List(("chun",21),("chun1",23),("chun3",22)))//隐士转换（RDD转换DF/DS需要引入隐式转换）import spark.implicits._//  rdd转DFval frame = rdd1.toDF("name","age")//创建全局视图frame.createGlobalTempView("people")//创建聚合函数对象val udaf = new MyAgeFunction//注册聚合函数spark.udf.register("avgAge",udaf)//frame.select("age").show()//sql  这里表名要把全局名也写上spark.sql("select avgAge(age) from global_temp.people").show}

2）创建强类型聚合函数AVG（extends Aggregator[输入类型,缓冲区类型,输出类型]）


//声明自定义聚合函数（强类型）
//case class Aggregator[K, V, C] (这里由三个泛型)
class MyAgeClassFuction extends Aggregator[UserBean,AvgBuffer,Double]{//初始化缓冲区override def zero: AvgBuffer = AvgBuffer(0,0)//AvgBuffer =  把输入的数据更新进缓冲区override def reduce(b: AvgBuffer, a: UserBean): AvgBuffer = {//sum和count要设置为var的b.sum += a.ageb.count += 1b}//合并缓冲区override def merge(b1: AvgBuffer, b2: AvgBuffer): AvgBuffer = {b1.sum = b1.sum + b2.sumb1.count = b1.count + b2.countb1}//计算结果override def finish(reduction: AvgBuffer): Double = {reduction.sum / reduction.count}//后俩都是数据变成类型之后的转码操作//第一个是自定义的类型，就用Encoders.productoverride def bufferEncoder: Encoder[AvgBuffer] = Encoders.product//如果不是自定义类型就用Encoders.scalaBooleanoverride def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}//样例类
case class UserBean(name : String, age : Int)
case class AvgBuffer(var sum : Int, var count : Int)

使用

  def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("Spark02_Custom2").setMaster("local[*]")val spark = SparkSession.builder().config(conf).getOrCreate()val rdd = spark.sparkContext.makeRDD(List(("chun1",23),("chun2",24),("chun3",25)))import spark.implicits._rdd.toDF("name","age")//自定义强类型聚合函数val udaf = new MyAgeClassFuction//这里不能注册，加入注册了名为avgAge，使用的时候是avgAge（字段），但是传入的应该是Bean对象，所以不可以这样写//需要将聚合函数转换为查询列val avgColumn = udaf.toColumn.name("avgAge")val userRDD = rdd.map {case (name, age) => {UserBean(name, age)}}//在sql里肯定没办法用，需要使用DSL风格select函数val ds = userRDD.toDSval rdd1 = ds.rddds.show()/****结果：+-----+---+| name|age|+-----+---+|chun1| 23||chun2| 24||chun3| 25|+-----+---+**/rdd1.foreach(println)//结果：//UserBean(chun1,23)//UserBean(chun3,25)//UserBean(chun2,24)spark.stop()}