json数据

  1. [{"name":"张三" ,"age":18} ,{"name":"李四" ,"age":15}]

代码编写

  1. import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
  2. import org.apache.spark.sql.types._
  3. import org.apache.spark.sql.{Row, SparkSession}
  5. import scala.collection.immutable.Nil
  7. object UDAFDemo {
  8.   def main(args: Array[String]): Unit = {
  9.     // 在sql中, 聚合函数如何使用
  10.     val spark: SparkSession = SparkSession
  11.       .builder()
  12.       .master("local[*]")
  13.       .appName("UDAFDemo")
  14.       .getOrCreate()
  15.     val df = spark.read.json("E:\\ZJJ_SparkSQL\\demo01\\src\\main\\resources\\users.json")
  16.     df.createOrReplaceTempView("user")
  17.     // 注册聚合函数
  18.     spark.udf.register("myAvg", new MyAvg)
  19.     spark.sql("select myAvg(age) from user").show
  20.     spark.close()
  21.   }
  22. }
  24. /**
  25.  * 求平均值
  26.  */
  27. class MyAvg extends UserDefinedAggregateFunction {
  28.   // 输入的数据类型  10.1 12.2 100
  29.   override def inputSchema: StructType = StructType(StructField("ele", DoubleType) :: Nil)
  31.   // 缓冲区的类型
  32.   // 求平均值需要两个值运算,一个是年龄的和,另外一个是多少个年龄参与运算,
  33.   // 所以这就是求平均值了.
  34.   override def bufferSchema: StructType =
  35.     StructType(StructField("sum", DoubleType) :: StructField("count", LongType) :: Nil)
  37.   // 最终聚合解结果的类型
  38.   override def dataType: DataType = DoubleType
  40.   // 相同的输入是否返回相同的输出
  41.   override def deterministic: Boolean = true
  43.   // 对缓冲区初始化
  44.   override def initialize(buffer: MutableAggregationBuffer): Unit = {
  45.     // avg初始化是一个值,个数初始化也得是一个值.
  47.     // 在缓冲集合中初始化和
  48.     buffer(0) = 0D // 等价于 buffer.update(0, 0D)
  49.     buffer(1) = 0L //带个L,不然就存成int类型了.
  50.   }
  52.   // 分区内聚合
  53.   override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
  54.     input match {
  55.       case Row(age: Double) =>
  56.         buffer(0) = buffer.getDouble(0) + age //年龄进行相加
  57.         buffer(1) = buffer.getLong(1) + 1L //个数累加碰到一个就加1
  58.       case _ =>
  59.     }
  61.     /*// input是指的使用聚合函数的时候, 缓过来的参数封装到了Row
  62.     if (!input.isNullAt(0)) { // 考虑到传字段可能是null
  63.         val v = input.getAs[Double](0) // getDouble(0)
  64.         buffer(0) = buffer.getDouble(0) + v
  65.         buffer(1) = buffer.getLong(1) + 1L
  66.     }*/
  67.   }
  69.   // 分区间的聚合
  70.   override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
  71.     buffer2 match {
  72.       case Row(sum: Double, count: Long) =>
  73.         // 缓冲区和要集合
  74.         buffer1(0) = buffer1.getDouble(0) + sum
  75.         //个数也要聚合
  76.         buffer1(1) = buffer1.getLong(1) + count
  78.       case _ =>
  79.     }
  82.     // 把buffer1和buffer2 的缓冲弄聚合到一起, 然后再把值写回到buffer1
  83.     /*buffer1(0) = buffer1.getDouble(0) + buffer2.getDouble(0)
  84.     buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)*/
  85.   }
  87.   // 返回最终的输出值
  88.   // 就是累加的总数除以个数,就是平均值了.
  89.   override def evaluate(buffer: Row): Any = buffer.getDouble(0) / buffer.getLong(1)
  90. }
  92. /**
  93.  * 求和的函数
  94.  */
  95. class MySum extends UserDefinedAggregateFunction {
  96.   // 输入的数据类型  10.1 12.2 100
  97.   override def inputSchema: StructType = StructType(StructField("ele", DoubleType) :: Nil)
  99.   // 缓冲区的类型,求和的时候需要缓冲,计算聚合的是一定要有缓冲
  100.   override def bufferSchema: StructType = StructType(StructField("sum", DoubleType) :: Nil)
  102.   // 最终聚合解结果的类型
  103.   //因为你是聚合,聚合的结果只有一个
  104.   override def dataType: DataType = DoubleType
  106.   // 相同的输入是否返回相同的输出
  107.   //用的时候几乎永远都是true
  108.   override def deterministic: Boolean = true
  110.   // 对缓冲区初始化
  111.   //初始化的就是在缓冲里面去初始化一个值,用来计算聚合的值的.
  112.   override def initialize(buffer: MutableAggregationBuffer): Unit = {
  113.     // 在缓冲集合中初始化和
  114.     buffer(0) = 0D // 等价于 buffer.update(0, 0D)
  115.   }
  117.   // 分区内聚合
  118.   override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
  119.     // 需要先进行非空判断.
  120.     // 如果不为空的话就取出来进行计算.
  122.     input match {
  123.       case Row(age: Double) =>
  124.         // 获取0位置的
  125.         buffer(0) = buffer.getDouble(0) + age
  127.       case _ =>
  128.     }
  130.     /*// input是指的使用聚合函数的时候, 缓过来的参数封装到了Row
  131.     if (!input.isNullAt(0)) { // 考虑到传字段可能是null
  132.         val v = input.getAs[Double](0) // getDouble(0)
  133.         buffer(0) = buffer.getDouble(0) + v
  134.     }*/
  135.   }
  137.   // 分区间的聚合
  138.   override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
  139.     // 把buffer1和buffer2 的缓冲弄聚合到一起, 然后再把值写回到buffer1
  140.     //这里不需要判断非空,因为缓冲区初始化(initialize方法)是一定有值的.
  141.     buffer1(0) = buffer1.getDouble(0) + buffer2.getDouble(0)
  143.   }
  145.   // 集合函数返回最终的输出值
  146.   override def evaluate(buffer: Row): Any = buffer.getDouble(0)
  147. }
  1. +--------------------------+
  2. |myavg(CAST(age AS DOUBLE))|
  3. +--------------------------+
  4. |                      16.5|
  5. +--------------------------+