UDF

计算平均年龄

  1. {"username": "zs" , "age" :  18}
  2. {"username": "ls" , "age" :  19}
  1.   case class Buff( var valCount : Long , var indexCount : Long)
  3.   class MyAvgUDAF extends Aggregator[ Long, Buff , Long ] {  // [输入值，缓冲区，输出值]
  4.       // 缓冲区初始值
  5.       override def zero: Buff = {
  6.          Buff(0L,0L)
  7.       }
  8.       // 输入值 到 缓冲区聚合
  9.       override def reduce(buff: Buff, a: Long): Buff = {
  10.         buff.valCount += a
  11.         buff.indexCount += 1
  12.         buff
  13.       }
  14.       // 缓冲区 分区间 聚合
  15.       override def merge(b1: Buff, b2: Buff): Buff = {
  16.         b1.valCount += b2.valCount
  17.         b1.indexCount += b2.indexCount
  18.         b1
  19.       }
  20.       //最终输出
  21.       override def finish(res: Buff): Long = {
  22.         res.valCount / res.indexCount
  23.       }
  25.       override def bufferEncoder: Encoder[Buff] = Encoders.product   // 如果是自定义类型 就必须这样
  27.       override def outputEncoder: Encoder[Long] = Encoders.scalaLong // 如果不是 看类型
  28.   }
  29. }
  1. def main(args: Array[String]): Unit = {
  3.       //TODO 创建SparkSQL的运行环境
  4.       val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL_UDAF1")
  5.       val spark = SparkSession.builder().config(sparkConf).getOrCreate()
  7.       //TODO 执行逻辑操作
  8.       val df: DataFrame = spark.read.json("datas/sql/user.json")
  9.       df.show()
  11.       //用户自定义聚合函数
  12.       spark.udf.register("userAvg", functions.udaf(new MyAvgUDAF))
  14.       df.createOrReplaceTempView("user")
  15.       spark.sql("select userAvg(age) from user").show()
  18.       //TODO 关闭环境
  19.       spark.close()
  20.   }