版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
传送门:大数据系列文章目录
官方网址:http://spark.apache.org/、 http://spark.apache.org/sql/
无论Hive还是SparkSQL分析处理数据时,往往需要使用函数, SparkSQL模块本身自带很多实现公共功能的函数,在org.apache.spark.sql.functions中。 SparkSQL与Hive一样支持定义函数: UDF和UDAF,尤其是UDF函数在实际项目中使用最为广泛。
回顾Hive中自定义函数有三种类型:
第一种: UDF(User-Defined-Function) 函数
第二种: UDAF(User-Defined Aggregation Function) 聚合函数
第三种: UDTF(User-Defined Table-Generating Functions) 函数
目前来说Spark 框架各个版本及各种语言对自定义函数的支持:
在SparkSQL中,目前仅仅支持UDF函数和UDAF函数:
由于SparkSQL数据分析有两种方式: DSL编程和SQL编程,所以定义UDF函数也有两种方式,不同方式可以在不同分析中使用。
使用SparkSession中udf方法定义和注册函数,在SQL中使用,使用如下方式定义:
使用org.apache.sql.functions.udf函数定义和注册函数,在DSL中使用,如下方式:
测试数据
{"name":"Michael", "salary":3000}
{"name":"Andy", "salary":4500}
{"name":"Justin", "salary":3500}
{"name":"Berta", "salary":4000}
范例演示: 将姓名转换为小写,调用String中toLowerCase方法。
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
* SparkSQL中定义UDF函数,两种方式,需求:将字符串转换为小写字母
*/
object SparkSQLUdf {
def main(args: Array[String]): Unit = {
// 构建SparkSession实例对象
val spark: SparkSession = SparkSession.builder()
.master("local[4]")
.appName(this.getClass.getSimpleName.stripSuffix("$"))
// TODO: 设置shuffle时分区数目
.config("spark.sql.shuffle.partitions", "4")
.getOrCreate()
Logger.getRootLogger.setLevel(Level.WARN)
// 导入隐式转换
import spark.implicits._
// 导入函数库
import org.apache.spark.sql.functions._
// 读取JSON格式数据
val empDF: DataFrame = spark.read.json("datas/resources/employees.json")
/*
root
|-- name: string (nullable = true)
|-- salary: long (nullable = true)
*/
empDF.printSchema()
/*
+-------+------+
|name |salary|
+-------+------+
|Michael|3000 |
|Andy |4500 |
|Justin |3500 |
|Berta |4000 |
+-------+------+
*/
empDF.show(10, truncate = false)
// TODO: 定义UDF函数,在SQL中使用
spark.udf.register(
"lower_name", // 函数名称
(name: String) => name.toLowerCase
)
// 注册DataFrame为临时视图
empDF.createOrReplaceTempView("view_tmp_emp")
spark.sql(
"""
|SELECT name, lower_name(name) AS new_name FROM view_tmp_emp
""".stripMargin)
.show(10, truncate = false)
println("==================================================")
// TODO: 定义UDF函数,在DSL中使用
val lower_udf: UserDefinedFunction = udf(
(name: String) => name.toLowerCase
)
empDF
.select(
$"name", //
lower_udf($"name").as("new_name") //
)
.show(10, truncate = false)
// 应用结束,关闭资源
spark.stop()
}
}
执行结果:
root
|-- name: string (nullable = true)
|-- salary: long (nullable = true)
+-------+------+
|name |salary|
+-------+------+
|Michael|3000 |
|Andy |4500 |
|Justin |3500 |
|Berta |4000 |
+-------+------+
+-------+--------+
|name |new_name|
+-------+--------+
|Michael|michael |
|Andy |andy |
|Justin |justin |
|Berta |berta |
+-------+--------+
==================================================
+-------+--------+
|name |new_name|
+-------+--------+
|Michael|michael |
|Andy |andy |
|Justin |justin |
|Berta |berta |
+-------+--------+