Spark SQL自定义函数 1

简介: Spark SQL自定义函数
+关注继续查看

1 Spark SQL自定义函数

1.1 自定义函数分类

类似于hive当中的自定义函数, spark同样可以使用自定义函数来实现新的功能。

spark中的自定义函数有如下3类

image

1.UDF(User-Defined-Function)

输入一行,输出一行

2.UDAF(User-Defined Aggregation Funcation)

输入多行,输出一行

3.UDTF(User-Defined Table-Generating Functions)

输入一行,输出多行

image

1.2 自定义UDF

image

image

image

image

image

image

image

image

●需求

有udf.txt数据格式如下:

Hello
abc
study
small

通过自定义UDF函数将每一行数据转换成大写

select value,smallToBig(value) from t_word

●代码演示

package cn.oldlu.sql

import org.apache.spark.SparkContext
import org.apache.spark.sql.{Dataset, SparkSession}


object UDFDemo {
  def main(args: Array[String]): Unit = {
    //1.创建SparkSession
    val spark: SparkSession = SparkSession.builder().master("local[*]").appName("SparkSQL").getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //2.读取文件
    val fileDS: Dataset[String] = spark.read.textFile("D:\\data\\udf.txt")
    fileDS.show()
    /*
    +----------+
    |     value|
    +----------+
    |helloworld|
    |       abc|
    |     study|
    | smallWORD|
    +----------+
     */
   /*
    将每一行数据转换成大写
    select value,smallToBig(value) from t_word
    */
    //注册一个函数名称为smallToBig,功能是传入一个String,返回一个大写的String
    spark.udf.register("smallToBig",(str:String) => str.toUpperCase())
    fileDS.createOrReplaceTempView("t_word")
    //使用我们自己定义的函数
    spark.sql("select value,smallToBig(value) from t_word").show()
    /*
    +----------+---------------------+
    |     value|UDF:smallToBig(value)|
    +----------+---------------------+
    |helloworld|           HELLOWORLD|
    |       abc|                  ABC|
    |     study|                STUDY|
    | smallWORD|            SMALLWORD|
    +----------+---------------------+
     */
    sc.stop()
    spark.stop()
  }
}

1.3 自定义UDAF

虽然小编说UDAF只做了解,但也只需要掌握的。在企业中,这也是你和别人拉开差距的地方

●需求

有udaf.json数据内容如下

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

求取平均工资

●继承UserDefinedAggregateFunction方法重写说明

image

image

inputSchema:输入数据的类型

image

bufferSchema:产生中间结果的数据类型

image

dataType:最终返回的结果类型

image

deterministic:确保一致性,一般用true

image

initialize:指定初始值

image

update:每有一条数据参与运算就更新一下中间结果(update相当于在每一个分区中的运算)

image

merge:全局聚合(将每个分区的结果进行聚合)

image

evaluate:计算最终的结果

image

image

●代码演示

package cn.oldlu.sql

import org.apache.spark.SparkContext
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Row, SparkSession}


object UDAFDemo {
  def main(args: Array[String]): Unit = {
    //1.获取sparkSession
    val spark: SparkSession = SparkSession.builder().appName("SparkSQL").master("local[*]").getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //2.读取文件
    val employeeDF: DataFrame = spark.read.json("D:\\data\\udaf.json")
    //3.创建临时表
    employeeDF.createOrReplaceTempView("t_employee")
    //4.注册UDAF函数
    spark.udf.register("myavg",new MyUDAF)
    //5.使用自定义UDAF函数
    spark.sql("select myavg(salary) from t_employee").show()
    //6.使用内置的avg函数
    spark.sql("select avg(salary) from t_employee").show()
  }
}
class MyUDAF extends UserDefinedAggregateFunction{
  //输入的数据类型的schema
  override def inputSchema: StructType = {
     StructType(StructField("input",LongType)::Nil)
  }
  //缓冲区数据类型schema,就是转换之后的数据的schema
  override def bufferSchema: StructType = {
    StructType(StructField("sum",LongType)::StructField("total",LongType)::Nil)
  }
  //返回值的数据类型
  override def dataType: DataType = {
    DoubleType
  }
  //确定是否相同的输入会有相同的输出
  override def deterministic: Boolean = {
    true
  }
  //初始化内部数据结构
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0L
    buffer(1) = 0L
  }
  //更新数据内部结构,区内计算
  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)
  }
}


目录
相关文章
|
4天前
|
SQL 分布式计算 数据可视化
Spark SQL案例【电商购买数据分析】
Spark SQL案例【电商购买数据分析】
|
4天前
|
SQL 分布式计算 数据可视化
Spark SQL【基于泰坦尼克号生还数据的 Spark 数据分析处理】
Spark SQL【基于泰坦尼克号生还数据的 Spark 数据分析处理】
|
4天前
|
SQL 分布式计算 关系型数据库
Spark【Spark SQL(二)RDD转换DataFrame、Spark SQL读写数据库 】
Spark【Spark SQL(二)RDD转换DataFrame、Spark SQL读写数据库 】
|
1月前
|
SQL 存储 分布式计算
pyspark笔记(RDD,DataFrame和Spark SQL)2
pyspark笔记(RDD,DataFrame和Spark SQL)
25 2
|
1月前
|
SQL 分布式计算 HIVE
pyspark笔记(RDD,DataFrame和Spark SQL)1
pyspark笔记(RDD,DataFrame和Spark SQL)
29 1
|
1月前
|
SQL 分布式计算 Java
201 Spark SQL查询程序
201 Spark SQL查询程序
21 0
|
1月前
|
SQL 分布式计算 Spark
200 Spark DataFrame常用操作- SQL风格语法
200 Spark DataFrame常用操作- SQL风格语法
13 0
|
2月前
|
SQL Oracle 关系型数据库
使用PL/SQL&自定义函数(Oracle)
使用PL/SQL&自定义函数(Oracle)
|
3月前
|
SQL 分布式计算 Spark
Spark SQL自定义函数 2
Spark SQL自定义函数
29 0
推荐文章
更多