Spark SQL自定义函数

1.自定义函数分类
类似于hive当中的自定义函数， spark同样可以使用自定义函数来实现新的功能。
spark中的自定义函数有如下3类
1.UDF(User-Defined-Function)
输入一行，输出一行
2.UDAF(User-Defined Aggregation Funcation)
输入多行，输出一行
3.UDTF(User-Defined Table-Generating Functions)
输入一行，输出多行
2、自定义UDF
需求
有udf.txt数据格式如下：

Hello
abc
study
small

通过自定义UDF函数将每一行数据转换成大写
select value,smallToBig(value) from t_word
代码演示:

def main(args: Array[String]): Unit = {
//1、创建sparksession
val spark: SparkSession = SparkSession.builder().master(“local[*]”).appName(“demo01”).getOrCreate()
//2、创建sparkcontext
val sc: SparkContext = spark.sparkContext
//3、读取数据。并操作
val ttRDD: RDD[String] = sc.textFile(“file:///F:\传智播客\传智专修学院\第二学期\34\05-Spark\资料\udf.txt”)
import spark.implicits._
val UDFDS: Dataset[String] = ttRDD.toDS()
//自定义函数
spark.udf.register(“toUpperAdd123”,(str:String)=>{
//根据业务需求对数据进行加工
str.toUpperCase +" 123"
})
UDFDS.createOrReplaceTempView(“UDF”)
//调用函数
spark.sql(“SELECT value,toUpperAdd123(value) as length_10 FROM UDF”).show()
sc.stop()
spark.stop()
}

3、自定义UDAF[了解]
需求：
有udaf.json数据内容如下

{“name”:“Michael”,“salary”:3000}
{“name”:“Andy”,“salary”:4500}
{“name”:“Justin”,“salary”:3500}
{“name”:“Berta”,“salary”:4000}

求取平均工资
●继承UserDefinedAggregateFunction方法重写说明
inputSchema：输入数据的类型
bufferSchema：产生中间结果的数据类型
dataType：最终返回的结果类型
deterministic：确保一致性，一般用true
initialize：指定初始值
update：每有一条数据参与运算就更新一下中间结果(update相当于在每一个分区中的运算)
merge：全局聚合(将每个分区的结果进行聚合)
evaluate：计算最终的结果
●代码演示:

object Udaf {
//编写计算平均工资的方法SalaryAvg
class SalaryAvg extends UserDefinedAggregateFunction{
//输入的数据的类型
override def inputSchema: StructType = {
StructType(StructField(“input”,LongType)::Nil)
}
//中间结果缓存的类型
override def bufferSchema: StructType = {
//sum缓存总金额
//total缓存总次数
StructType(List(StructField(“sum”,LongType),(StructField(“total”,LongType))))
}
//数据返回的类型
override def dataType: DataType = {
DoubleType
}
//是否有相同的输出 true
override def deterministic: Boolean = {
true
}
/*
List(1,2,3,4,5).reduce（(a,b)=>a+b）
1 a=1 b=2
2 a=3 b=3
3 a=6 b=4
4 a=10 b=5
5 a=51
/
//数据的初始化
override def initialize(buffer: MutableAggregationBuffer): Unit = {
//用于存储总金额
buffer(0)=0L //=> a
//用于存储次数
buffer(1)=0L //=>b
}
//rdd是分区的此方法是计算一个分区内的数据和和数据数量
/
{“name”:“Michael”,“salary”:3000}
{“name”:“Andy”,“salary”:4500}
{“name”:“Justin”,“salary”:3500}
{“name”:“Berta”,“salary”:4000}
/
override def update(buffer: MutableAggregationBuffer, input: Row): Unit ={
//计算次分区的总金额
buffer(0)=buffer.getLong(0)+input.getLong(0)
//计算次分区的总数量
buffer(1)=buffer.getLong(1)+1
}
//汇总所有分区内的总金额和总次数
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): Any = {
buffer.getLong(0).toDouble/buffer.getLong(1).toDouble
}
}
def main(args: Array[String]): Unit = {
//1、创建sparksession
val spark: SparkSession = SparkSession.builder().master("local[]").appName(“demo01”).getOrCreate()
val JsonDatas: DataFrame = spark.read.json(“file:///F:\传智播客\传智专修学院\第二学期\34\05-Spark\资料\udaf.json”)
JsonDatas.createOrReplaceTempView(“UDAFTable”)
//注册程UDAF函数
spark.udf.register(“SalaryAvg”,new SalaryAvg)
//计算平均工资的算法名称为SalaryAvg
spark.sql(“select SalaryAvg(salary) from UDAFTable”).show()
spark.sql(“select avg(salary) from UDAFTable”).show()
spark.stop()
}
}

Spark SQL自定义函数_第五章相关推荐

spark SQL自定义函数:
spark SQL 自定义函数: 自定义函数: 第一种: U D F (用户自定义函数)函数特点: 一对一的关系,输入一个值以后输出一个值 (一进一出) 大部分的内置函数都是U D F函数 ...
Spark SQL自定义函数
文章目录自定义函数分类自定义UDF 自定义UDAF[了解] 自定义函数分类类似于hive当中的自定义函数, spark同样可以使用自定义函数来实现新的功能. spark中的自定义函数有如下3类 ...
spark sql自定义UDF函数-java语言
背景说明基于spark sql开发过程中,需要一些类似与官网提供的 int().from_json()等自定函数处理数据.下属将简单讲解通过java如何实现spark sql自定义函数官方UDF接 ...
sql自定义函数学习思路_学习SQL：用户定义的函数
sql自定义函数学习思路 You can create several user-defined objects in a database. One of these is definitely u ...
sql 自定义函数示例_SQL滞后函数概述和示例
sql 自定义函数示例 In the article SQL Server Lead function overview and examples, we explored Lead functio ...
ArcGIS for Desktop入门教程_第五章_ArcCatalog使用 - ArcGIS知乎-新一代ArcGIS问答社区
原文:ArcGIS for Desktop入门教程_第五章_ArcCatalog使用 - ArcGIS知乎-新一代ArcGIS问答社区 1 ArcCatalog使用 1.1 GIS数据地理信息系统, ...
sql 自定义函数示例_SQL Server Choose（）函数介绍和示例
sql 自定义函数示例 In the article, a CASE statement in SQL, we explored one of the important logical expre ...
MS SQL自定义函数IsPositiveInteger MS SQL自定义函数IsNumeric 水晶报表使用IEnumerableT数据源...
MS SQL自定义函数IsPositiveInteger 判断字符串是否为正整数,0开始的的数字不算. SET ANSI_NULLS ON GO SET QUOTED_IDENTIFIER ON GO ...
sql 自定义函数示例_SQL Server SESSION_CONTEXT（）函数与示例
sql 自定义函数示例 This article explores the SQL Server session context function, SESSION_CONTEXT() and pe ...

Spark SQL自定义函数_第五章

Spark SQL自定义函数_第五章相关推荐

最新文章

热门文章