第三章第六節 Spark-SQL核心編程（五）自定義函數：UDF：val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SQLDemo")//創建SparkSession對象val spark :SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()import spark.implicits._//讀取json文件val df : DataFrame = spark.read.json("Spark-SQL/input/user.json")spark.udf.register("addName",(x:String)=>"Name:"+x)df.createOrReplaceTempView("people")spark.sql("select addName(username),age from people").show()spark.stop()UDAF（自定義聚合函數）強類型的 Dataset 和弱類型的 DataFrame 都提供了相關的聚合函數， 如 count()，countDistinct()，avg()，max()，min()。除此之外，用戶可以設定自己的自定義聚合函數。Spark3.0之前我們使用的是UserDefinedAggregateFunction作為自定義聚合函數，從 Spark3.0 版本后可以統一采用強類型聚合函數 Aggregator實驗需求：計算平均工資實現方式一：RDDval sparkconf: SparkConf = new SparkConf().setAppName("app").setMaster("local[*]")val sc: SparkContext = new SparkContext(conf)val resRDD: (Int, Int) = sc.makeRDD(List(("zhangsan", 20), ("lisi", 30), ("wangwu",40))).map { case (name, salary) =>{ (salary, 1) }}.reduce { (t1, t2) =>{ (t1._1 + t2._1, t1._2 + t2._2) }}println(resRDD._1/resRDD._2)// 關閉連接sc.stop()實現方式二：弱類型UDAFclass MyAverageUDAF extends UserDefinedAggregateFunction{ def inputSchema: StructType = StructType(Array(StructField("salary",IntegerType))) // 聚合函數緩沖區中值的數據類型(salary,count) def bufferSchema: StructType = { StructType(Array(StructField("sum",LongType),StructField("count",LongType))) } // 函數返回值的數據類型 def dataType: DataType = DoubleType // 穩定性：對于相同的輸入是否一直返回相同的輸出。 def deterministic: Boolean = true // 函數緩沖區初始化 def initialize(buffer: MutableAggregationBuffer): Unit = { // 存薪資的總和 buffer(0) = 0L // 存薪資的個數 buffer(1) = 0L } // 更新緩沖區中的數據 def update(buffer: MutableAggregationBuffer,input: Row): Unit = { if (!input.isNullAt(0)) { buffer(0) = buffer.getLong(0) + input.getInt(0) buffer(1) = buffer.getLong(1) + 1 } } // 合并緩沖區 def merge(buffer1: MutableAggregationBuffer,buffer2: Row): Unit = { buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0) buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1) } // 計算最終結果 def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1)}val sparkconf: SparkConf = new SparkConf().setAppName("app").setMaster("local[*]")val spark:SparkSession = SparkSession.builder().config(conf).getOrCreate()import spark.implicits._val res :RDD[(String,Int)]= spark.sparkContext.makeRDD(List(("zhangsan", 20), ("lisi", 30), ("wangwu",40)))val df :DataFrame = res.toDF("name","salary")df.createOrReplaceTempView("user")var myAverage = new MyAverageUDAF//在 spark 中注冊聚合函數spark.udf.register("avgSalary",myAverage)spark.sql("select avgSalary(salary) from user").show()// 關閉連接spark.stop()實現方式三：強類型UDAFcase class Buff(var sum:Long,var cnt:Long)class MyAverageUDAF extends Aggregator[Long,Buff,Double]{ override def zero: Buff = Buff(0,0) override def reduce(b: Buff, a: Long): Buff = { b.sum += a b.cnt += 1 b } override def merge(b1: Buff, b2: Buff): Buff = { b1.sum += b2.sum b1.cnt += b2.cnt b1 } override def finish(reduction: Buff):?