• Spark【Spark SQL（四）UDF函数和UDAF函数】

    UDF 函数

            UDF 是我们用户可以自定义的函数，我们通过SparkSession对象来调用 udf 的 register（name：String，func（A1,A2,A3...）） 方法来注册一个我们自定义的函数。其中，name 是我们自定义的函数名称，func 是我们自定义的函数，它可以有很多个参数。

            通过 UDF 函数，我们可以针对某一列数据或者某单元格数据进行针对的处理。

    案例 1

    定义一个函数，给 Andy 的 name 字段的值前 + "Name: "。

    1. def main(args: Array[String]): Unit = {
    2.  
    3.     val conf = new SparkConf()
    4.     conf.setAppName("spark sql udf")
    5.       .setMaster("local[*]")
    6.     val spark = SparkSession.builder().config(conf).getOrCreate()
    7.     import spark.implicits._
    8.  
    9.     val df = spark.read.json("data/sql/people.json")
    10.     df.createOrReplaceTempView("people")
    11.  
    12.     spark.udf.register("prefixName",(name:String)=>{
    13.       if (name.equals("Andy"))
    14.         "Name: " + name
    15.       else
    16.         name
    17.     })
    18.     spark.sql("select prefixName(name) as name,age,sex from people").show()
    19.  
    20.     spark.stop()
    21.   }

            这里我们定义了一个自定义的 UDF 函数：prefixName，它会判断name字段的值是否为 "Andy"，如果是，就会在她的值前+"Name: "。

    运行结果：

    1. +----------+---+---+
    2. |      name|age|sex|
    3. +----------+---+---+
    4. |   Michael| 30| 男|
    5. |Name: Andy| 19| 女|
    6. |    Justin| 19| 男|
    7. |Bernadette| 20| 女|
    8. |  Gretchen| 23| 女|
    9. |     David| 27| 男|
    10. |    Joseph| 33| 女|
    11. |     Trish| 27| 女|
    12. |      Alex| 33| 女|
    13. |       Ben| 25| 男|
    14. +----------+---+---+

    UDAF 函数

            强类型的DataSet和弱类型的DataFrame都提供了相关聚合函数，如count、countDistinct、avg、max、min。

            UDAF 也就是我们用户的自定义聚合函数。聚合函数就比如 avg、sum这种函数，需要先把所有数据放到一起（缓冲区），再进行统一处理的一个函数。

            实现 UDAF 函数需要有我们自定义的聚合函数的类（主要任务就是计算），我们可以继承 UserDefinedAggregateFunction，并实现里面的八种方法，来实现弱类型的聚合函数。（Spark3.0之后就不推荐使用了，更加推荐强类型的聚合函数）

            我们可以继承Aggregator来实现强类型的聚合函数。

    案例1 - 平均年龄

    case 类可以直接构建对象，不需要new，因为样例类可以自动生成它的伴生对象和apply方法。

    弱类型实现

    1. import org.apache.spark.SparkConf
    2. import org.apache.spark.sql.{Row, SparkSession}
    3. import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
    4. import org.apache.spark.sql.types.{DataType, IntegerType, LongType, StructField, StructType}
    5.  
    6. /**
    7.  * 弱类型
    8.  */
    9. object UDAFTest01 {
    10.   def main(args: Array[String]): Unit = {
    11.     val conf = new SparkConf()
    12.     conf.setAppName("spark sql udaf")
    13.       .setMaster("local[*]")
    14.     val spark = SparkSession.builder().config(conf).getOrCreate()
    15.     import spark.implicits._
    16.  
    17.     val df = spark.read.json("data/sql/people.json")
    18.     df.createOrReplaceTempView("people")
    19.  
    20.     spark.udf.register("avgAge",new MyAvgUDAF())
    21.  
    22.     spark.sql("select avgAge(age) from people").show()
    23.  
    24.     spark.stop()
    25.   }
    26. }
    27. class MyAvgUDAF extends UserDefinedAggregateFunction{
    28.  
    29.   // 输入数据的结构 IN
    30.   override def inputSchema: StructType = {
    31.    StructType(
    32.      Array(StructField("age",LongType))
    33.    )}
    34.  
    35.   // 缓冲区数据的结构 BUFFER
    36.   override def bufferSchema: StructType = {
    37.     StructType(
    38.       Array(
    39.         StructField("total",LongType),
    40.         StructField("count",LongType)
    41.       )
    42.     )}
    43.  
    44.   // 函数计算结果的数据类型 OUT
    45.   override def dataType: DataType = LongType
    46.  
    47.   // 函数的稳定性 (传入相同的参数结果是否相同)
    48.   override def deterministic: Boolean = true
    49.  
    50.   // 缓冲区初始化
    51.   override def initialize(buffer: MutableAggregationBuffer): Unit = {
    52.     //这两种写法都一样
    53. //    buffer(0) = 0L
    54. //    buffer(1) = 0L
    55.     //第二种方法
    56.     buffer.update(0,0L) //total 给缓冲区的第0个数据结构-total-初始化赋值0L
    57.     buffer.update(1,0L) //count 给缓冲区的第1个数据结构-count-初始化赋值0L
    58.   }
    59.  
    60.   // 数据过来之后 如何更新缓冲区
    61.   override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    62.     // 第一个参数代表缓冲区的第i个数据结构 0代表total 1代表count
    63.     // 第二个参数是对第一个参数的数据结构进行重新赋值
    64.     // buffer.getLong(0)是取出缓冲区第0个值-也就是total的值,给它+上输入的值中的第0个值(因为我们输入结构只有一个就是age:Long)
    65.     buffer.update(0,buffer.getLong(0)+input.getLong(0))
    66.     buffer.update(1,buffer.getLong(1)+1)  //count 每次数据过来+1
    67.   }
    68.  
    69.   // 多个缓冲区数据合并
    70.   override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    71.     buffer1.update(0,buffer1.getLong(0)+buffer2.getLong(0))
    72.     buffer1.update(1,buffer1.getLong(1)+buffer2.getLong(1))
    73.   }
    74.  
    75.   // 计算结果操作
    76.   override def evaluate(buffer: Row): Any = {
    77.     buffer.getLong(0)/buffer.getLong(1)
    78.   }
    79. }

    运行结果：

    1. +-----------+
    2. |avgage(age)|
    3. +-----------+
    4. |         25|
    5. +-----------+

     

    强类型实现

    1. import org.apache.spark.SparkConf
    2. import org.apache.spark.sql.{Encoder, Encoders, Row, SparkSession, functions}
    3. import org.apache.spark.sql.expressions.Aggregator
    4.  
    5. /**
    6.  * 强类型
    7.  */
    8. object UDAFTest02 {
    9.   def main(args: Array[String]): Unit = {
    10.     val conf = new SparkConf()
    11.     conf.setAppName("spark sql udaf")
    12.       .setMaster("local[*]")
    13.     val spark = SparkSession.builder().config(conf).getOrCreate()
    14.     import spark.implicits._
    15.  
    16.     val df = spark.read.json("data/sql/people.json")
    17.     df.createOrReplaceTempView("people")
    18.  
    19.     spark.udf.register("avgAge",functions.udaf(new MyAvg_UDAF()))
    20.  
    21.     spark.sql("select avgAge(age) from people").show()
    22.  
    23.     spark.stop()
    24.   }
    25. }
    26.  
    27. /**
    28.  * 自定义聚合函数类：
    29.  *  1.继承org.apache.spark.sql.expressions.Aggregator,定义泛型:
    30.  *    IN  : 输入数据类型 Long
    31.  *    BUF : 缓冲区数据类型
    32.  *    OUT : 输出数据类型 Long
    33.  *  2.重写方法
    34.  */
    35. //样例类中的参数默认是 val 所以这里必须指定为var
    36. case class Buff(var total: Long,var count: Long)
    37. class MyAvg_UDAF extends Aggregator[Long,Buff,Long]{
    38.  
    39.   // zero: Buff zero代表这个方法是用来初始值(0值)
    40.   // Buff是我们的case类 也就是说明这里是用来给 缓冲区进行初始化
    41.   override def zero: Buff = {
    42.     Buff(0L,0L)
    43.   }
    44.  
    45.   // 根据输入数据更新缓冲区 要求返回-Buff
    46.   override def reduce(buff: Buff, in: Long): Buff = {
    47.     buff.total += in
    48.     buff.count += 1
    49.     buff
    50.   }
    51.  
    52.   // 合并缓冲区 同样返回buff1
    53.   override def merge(buff1: Buff, buff2: Buff): Buff = {
    54.     buff1.total += buff2.total
    55.     buff1.count += buff2.count
    56.     buff1
    57.   }
    58.  
    59.   // 计算结果
    60.   override def finish(buff: Buff): Long = {
    61.     buff.total/buff.count
    62.   }
    63.  
    64.   // 网络传输需要序列化 缓冲区的编码操作 -编码
    65.   override def bufferEncoder: Encoder[Buff] = Encoders.product
    66.  
    67.   // 输出的编码操作 -解码
    68.   override def outputEncoder: Encoder[Long] = Encoders.scalaLong
    69. }

    运行结果：

    1. +-----------+
    2. |avgage(age)|
    3. +-----------+
    4. |         25|
    5. +-----------+

     

    早期UDAF强类型聚合函数

    SQL：结构化数据查询 & DSL：面向对象查询（有对象有方法，与类型相关，所以通过DSL语句结合起来使用）

    早期的UDAF强类型聚合函数使用DSL操作。

    定义一个case类对应数据类型，然后通过as[对象]方法将DataFrame转为DataSet类型，然后将我们的UDAF聚合类转为列对象。

    1. import org.apache.spark.SparkConf
    2. import org.apache.spark.sql.{Dataset, Encoder, Encoders, Row, SparkSession, TypedColumn, functions}
    3. import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction}
    4. import org.apache.spark.sql.types.{DataType, IntegerType, LongType, StructField, StructType}
    5.  
    6. /**
    7.  * 早期的UDAF强类型聚合函数使用DSL操作
    8.  */
    9. object UDAFTest03 {
    10.  
    11.   def main(args: Array[String]): Unit = {
    12.     val conf = new SparkConf()
    13.     conf.setAppName("spark sql udaf")
    14.       .setMaster("local[*]")
    15.     val spark = SparkSession.builder().config(conf).getOrCreate()
    16.     import spark.implicits._
    17.  
    18.     val df = spark.read.json("data/sql/people.json")
    19.  
    20.     val ds: Dataset[User] = df.as[User]
    21.  
    22.     // 将UDAF强类型聚合函数转为查询的类对象
    23.     val udafCol: TypedColumn[User, Long] = new OldAvg_UDAF().toColumn
    24.     ds.select(udafCol).show()
    25.  
    26.     spark.stop()
    27.   }
    28. }
    29.  
    30. /**
    31.  * 自定义聚合函数类：
    32.  *  1.继承org.apache.spark.sql.expressions.Aggregator,定义泛型:
    33.  *    IN  : 输入数据类型 User
    34.  *    BUF : 缓冲区数据类型
    35.  *    OUT : 输出数据类型 Long
    36.  *  2.重写方法
    37.  */
    38. //样例类中的参数默认是 val 所以这里必须指定为var
    39. case class User(name: String,age: Long,sex: String)
    40. case class Buff(var total: Long,var count: Long)
    41. class OldAvg_UDAF extends Aggregator[User,Buff,Long]{
    42.  
    43.   // zero: Buff zero代表这个方法是用来初始值(0值)
    44.   // Buff是我们的case类 也就是说明这里是用来给 缓冲区进行初始化
    45.   override def zero: Buff = {
    46.     Buff(0L,0L)
    47.   }
    48.  
    49.   // 根据输入数据更新缓冲区 要求返回-Buff
    50.   override def reduce(buff: Buff, in: User): Buff = {
    51.     buff.total += in.age
    52.     buff.count += 1
    53.     buff
    54.   }
    55.  
    56.   // 合并缓冲区 同样返回buff1
    57.   override def merge(buff1: Buff, buff2: Buff): Buff = {
    58.     buff1.total += buff2.total
    59.     buff1.count += buff2.count
    60.     buff1
    61.   }
    62.  
    63.   // 计算结果
    64.   override def finish(buff: Buff): Long = {
    65.     buff.total/buff.count
    66.   }
    67.  
    68.   // 网络传输需要序列化 缓冲区的编码操作 -编码
    69.   override def bufferEncoder: Encoder[Buff] = Encoders.product
    70.  
    71.   // 输出的编码操作 -解码
    72.   override def outputEncoder: Encoder[Long] = Encoders.scalaLong
    73. }

    运行结果：

    1. +------------------------------------------+
    2. |OldAvg_UDAF(com.study.spark.core.sql.User)|
    3. +------------------------------------------+
    4. |                                        25|
    5. +------------------------------------------+

  • 相关阅读:
    Anaconda常用命令
    Docker版部署RocketMQ开启ACL验证
    利用ETLCloud自动化流程实现业务系统数据快速同步至数仓
    京东小程序：无代码开发实现API集成，连接电商平台、CRM和客服系统
    有关遗传算法最新发展的4篇论文推荐
    ＜Python＞PyQt5，多窗口之间参数传递和函数调用
    计算机毕业设计 高校课程评价系统的设计与实现 Java实战项目 附源码+文档+视频讲解
    简单讲解Android Fragment（四）
    【ubuntu】搭建lamp架构
    mnist数据集
  • 原文地址：https://blog.csdn.net/m0_64261982/article/details/132864647