目錄
一.UDF(一進一出)
步驟
① 注冊UDF函數，可以使用匿名函數。
② 在sql查詢的時候使用自定義的UDF。
示例
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
* @description: UDF一進一出
* @author: HaoWu
* @create: 2020年08月09日
*/
object UDF_Test {
def main(args: Array[String]): Unit = {
//創建SparkSession
val session: SparkSession = SparkSession.builder
.master("local[*]")
.appName("MyApp")
.getOrCreate()
//注冊UDF
session.udf.register("addHello",(name:String) => "hello:"+name)
//讀取json格式文件{"name":"zhangsan","age":20}，創建DataFrame
val df: DataFrame = session.read.json("input/1.txt")
//創建臨時視圖：person
df.createOrReplaceTempView("person")
//查詢的時候使用UDF
session.sql(
"""select
|addHello(name),
|age
|from person
|""".stripMargin).show
}
}
結果
|addHello(name)|age|
+--------------+---+
|hello:zhangsan| 20|
| hello:lisi| 30|
+--------------+---+
二.UDAF(多近一出)
spark2.X 實現方式
2.X版本：UserDefinedAggregateFunction 無類型或弱類型
步驟：
①繼承UserDefinedAggregateFunction，實現其中的方法
②創建函數對象，注冊函數，在sql中使用
//創建UDFA對象
val avgDemo1: Avg_UDAF_Demo1 = new Avg_UDAF_Demo1
//在spark中注冊聚合函數
spark.udf.register("ageDemo1", avgDemo1)
案例
需求：實現avg()聚合函數的功能，要求結果是Double類型
代碼實現
①繼承UserDefinedAggregateFunction，實現其中的方法
import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StructField, StructType}
/**
* @description: UDAF(多近一出):求age的平均值
* 2.X 版本繼承UserDefinedAggregateFunction類，弱類型
* 非常類似累加器，aggregateByKey算子的操作，有個ZeroValue，不斷將輸入的值做歸約操作，然后再賦值給ZeroValue
* @author: HaoWu
* @create: 2020年08月08日
*/
class Avg_UDAF_Demo1 extends UserDefinedAggregateFunction {
//聚合函數輸入參數的數據類型，
override def inputSchema = StructType(StructField("age", LongType) :: Nil)
//聚合函數緩沖區中值的數據類型(sum,count)
override def bufferSchema = StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)
//函數返回值的數據類型
override def dataType = DoubleType
//穩定性：對於相同的輸入是否一直返回相同的輸出，一般都是true
override def deterministic = true
//函數緩沖區初始化，就是ZeroValue清空
override def initialize(buffer: MutableAggregationBuffer): Unit = {
//緩存區看做一個數組，將每個元素置空
//sum
buffer(0) = 0L
//count
buffer(1) = 0L
}
//更新緩沖區中的數據->將輸入的值和緩存區數據合並
override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
//input是Row類型，通過getXXX(索引值)取數據
if (!input.isNullAt(0)) {
val age = input.getLong(0)
buffer(0) = buffer.getLong(0) + age
buffer(1) = buffer.getLong(1) + 1
}
}
//合並緩沖區 (sum1,count1) + (sum2,count2) 合並
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) = buffer.getLong(0).toDouble/buffer.getLong(1)
}
②創建函數對象，注冊函數，在sql中使用
/**
* @description: 實現集合函數avg的功能
* @author: HaoWu
* @create: 2020年08月13日
*/
object UDAF_Test {
def main(args: Array[String]): Unit = {
//創建SparkSession
val spark: SparkSession = SparkSession.builder
.master("local[*]")
.appName("MyApp")
.getOrCreate()
//讀取json格式文件{"name":"zhangsan","age":20}
val df: DataFrame = spark.read.json("input/1.txt")
//創建臨時視圖：person
df.createOrReplaceTempView("person")
//創建UDFA對象
val avgDemo1: Avg_UDAF_Demo1 = new Avg_UDAF_Demo1
//在spark中注冊聚合函數
spark.udf.register("ageDemo1", avgDemo1)
//查詢的時候使用UDF
spark.sql(
"""select
|ageDemo1(age)
|from person
|""".stripMargin).show
}
}
spark3.X實現方式
3.x版本： 認為2.X繼承UserDefinedAggregateFunction的方式過時，推薦繼承Aggregator ，是強類型
步驟：
①繼承Aggregator [-IN, BUF, OUT]，聲明泛型，實現其中的方法
abstract class Aggregator[-IN, BUF, OUT]
IN: 輸入的類型
BUF： 緩沖區類型
OUT: 輸出的類型
②創建函數對象，注冊函數，在sql中使用
//創建UDFA對象
val avgDemo2: Avg_UDAF_Demo2 = new Avg_UDAF_Demo2
//在spark中注冊聚合函數
spark.udf.register("myAvg",functions.udaf(avgDemo2))
注意：2.X和3.X的注冊方式不同
案例
需求：實現avg()聚合函數的功能，要求結果是Double類型
代碼實現
①繼承Aggregator [-IN, BUF, OUT]，聲明泛型，實現其中的方法
其中緩沖區數據用樣例類進行封裝。
MyBuffer類
/**
* 定義MyBuffer樣例類
* @param sum 組數據sum和
* @param count 組的數據個數
*/
case class MyBuffer(var sum: Long, var count: Long)
自定義UDAF函數
import org.apache.spark.sql.Encoders
import org.apache.spark.sql.expressions.Aggregator
/**
* @description: UDAF(多近一出):求age的平均值
* 3.X Aggregator，強類型
* 非常類似累加器，aggregateByKey算子的操作，有個ZeroValue，不斷將輸入的值做歸約操作，然后再賦值給ZeroValue
* @author: HaoWu
* @create: 2020年08月08日
*/
class Avg_UDAF_Demo2 extends Aggregator[Long, MyBuffer, Double] {
//函數緩沖區初始化，就是ZeroValue清空
override def zero = MyBuffer(0L, 0L)
//將輸入的值和緩存區數據合並
override def reduce(b: MyBuffer, a: Long) = {
b.sum = b.sum + a
b.count = b.count + 1
b
}
//合並緩沖區
override def merge(b1: MyBuffer, b2: MyBuffer) = {
b1.sum = b1.sum + b2.sum
b1.count = b1.count + b2.count
b1
}
//計算最終結果
override def finish(reduction: MyBuffer) = reduction.sum.toDouble / reduction.count
/* scala中
常見的數據類型： Encoders.scalaXXX
自定義的類型：ExpressionEncoder[T]() 返回 Encoder[T]
樣例類(都是Product類型)： Encoders.product[T]，返回Produce類型的Encoder!
*/
//緩存區的Encoder類型
override def bufferEncoder = Encoders.product[MyBuffer]
//輸出結果的Encoder類型
override def outputEncoder = Encoders.scalaDouble
}
②創建函數對象，注冊函數，在sql中使用
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.{DataFrame, Row, SparkSession, functions}
/**
* @description: 實現集合函數avg的功能
* @author: HaoWu
* @create: 2020年08月13日
*/
object UDAF_Test {
def main(args: Array[String]): Unit = {
//創建SparkSession
val spark: SparkSession = SparkSession.builder
.master("local[*]")
.appName("MyApp")
.getOrCreate()
//讀取json格式文件{"name":"zhangsan","age":20}
val df: DataFrame = spark.read.json("input/1.txt")
//創建臨時視圖：person
df.createOrReplaceTempView("person")
//創建UDFA對象
val avgDemo2: Avg_UDAF_Demo2 = new Avg_UDAF_Demo2
//在spark中注冊聚合函數
spark.udf.register("myAvg",functions.udaf(avgDemo2))
//查詢的時候使用UDF
spark.sql(
"""select
|myAvg(age)
|from person
|""".stripMargin).show
}
}