大数据学习18之Spark-SQL

1.概述

1.1.简介

        Spark SQL 是 Apache Spark 用于处理结构化数据的模块。

1.2.历史

1.2.1.Shark

        Hadoop诞生初期，Hive是唯一在Hadoop上运行的SQL-on-Hadoop工具，MR的中间计算过程产生了大量的磁盘落地操作，消耗了大量的I/O，降低了程序的运行效率。

        为了提高SQL-on-Hadoop的效率，大量的SQL-on-Hadoop工具被开发，Spark-SQL便是其中的一种，Spark-SQL的前身就是Shark。

1.2.2.Hive on Spark*

        Spark的开发较晚，那时候的主流数据仓库便是Hive，为了通用性，Spark就与Hive结合起来了，对于SQL语句的解析都交给Hive来进行处理，并且Spark程序一定程度上替代了Hive底层的MapReduce程序，提高了作业的计算效率。

1.2.3.Spark on Hive*

        随着Spark发展，Shark 对于 Hive 的太多依赖（如采用 Hive 的语法解析器、查询优化器等等）制约了 Spark的 One stack to rule them all 的既定方针，制约了 Spark 各个组件的相互集成，所以就提出了 SparkSQL 项目。

        并且Hive本身的迭代更新速度较慢，就算是现在的最新版本的Hive支持的Spark也才2.x.x,

同时Spark在3.0.0版本时做出了一系列的优化，如果还是依赖于与Hive的化Spark3.0以上的版本的是用不了的，那么Spark的优化就没有意义了。

        为了让Spark的优化变得可用，Spark就自己开发了一套用于SQL操作的模块，由之前的Shark来到了现在的Spark-SQL。

        经过这次的转变，Spark由原来的依赖Hive解析SQL变成了由自己的Spark-SQL模块解析的方式，但是保留了对Hive的元数据访问。

        也就是说，现在的Spark除了元数据外，几乎可以说是一个一栈式大数据框架了。

1.2.4.Hive on Spark vs. Spark on Hive

        Hive on Spark：Hive为主体，在Hive中继承Spark，Hive即存储元数据，也解析SQL语句，只是Hive将引擎从MR更换为Spark由 ，Spark 负责运算工作，但部署较为复杂。

        Spark on Hive：Spark为主体，Hive只负责元数据的存储，由Spark来解析和执行SQL语句，其中SQL语法为Spark-SQL语法，且部署简单。Spark on Hive 的优点在于它提供了更灵活的编程接口，适用于各种数据处理需求。

2.数据模型

2.1. RDD 和 DataFrame

2.1.1.RDD转DataFrame

//创建样例类
scala> case class User(id: Int, name: String, age: Int, gender: Int)
defined class User//创建 RDD
scala> val rdd = sc.makeRDD(List(User(1, "zhangsan", 18, 1), User(2, "lisi", 19, 0), User(3, "wangwu",20, 1)))//RDD 转 DataFrame
scala> val df = rdd.toDF

2.1.2.DataFrame 转 RDD

//创建 DataFrame
scala> val df = spark.read.json("file:///opt/spark-local/data/user/user.json")//DataFrame 转 RDD
scala> val rdd = df.rdd

 2.2.RDD 和 Dataset

2.2.1. RDD 转 Dataset

        RDD 和 Dataset 两个都是强类型模型，所以可以相互直接转换。

//创建样例类
scala> case class User(id: Int, name: String, age: Int, gender: Int)//创建 RDD
scala> val rdd = sc.makeRDD(List(User(1, "zhangsan", 18, 1), User(2, "lisi", 19, 0), User(3, "wangwu",20, 1)))//RDD 转 Dataset。
scala> val ds = rdd.toDS

2.2.2.Dataset 转 RDD

scala> val rdd = ds.rdd

2.3. DataFrame 和 Dataset

2.3.1.DataFrame 转 Dataset

        配合样例类使用 as[类型] 转换为 DataSet。

scala> val df = spark.read.json("file:///opt/yjx/spark-scalocal/data/user/user.json")scala> val ds = df.as[User]

2.3.2.Dataset 转 DataFrame

//创建 Dataset
scala> case class User(id: Int, name: String, age: Int, gender: Int)defined class Userscala> val list = List(User(1, "zhangsan", 18, 1), User(2, "lisi", 19, 0), User(3, "wangwu", 20, 1))
list: List[User] = List(User(1,zhangsan,18,1), User(2,lisi,19,0), User(3,wangwu,20,1))scala> val ds = list.toDS//Dataset 转 DataFrame
scala> val df = ds.toDF

3. IDEA 开发 SparkSQL

        创建普通 Maven 项目，添加以下依赖。

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.3.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.3.2</version>
</dependency>

3.1.DataFrame

object DataFrameDemo {
case class User(id: Int, name: String, age: Int, gender: Int)
def main(args: Array[String]): Unit = {
// ==================== 建立连接 ====================
// 初始化配置对象并设置运行模式与 AppName
val conf = new SparkConf().setMaster("local[*]").setAppName("DataFrameDemo")
// 根据配置对象初始化 SparkSession 对象
val spark = SparkSession.builder().config(conf).getOrCreate()
// 日志级别
val sc = spark.sparkContext
sc.setLogLevel("ERROR")
// 在 IDEA 中开发 SparkSQL 如果遇到模型转换，需要导入隐式转换
import spark.implicits._
// ==================== 业务处理 ====================
// RDD 转 DataFrame
val rdd = sc.makeRDD(List(User(1, "zhangsan", 18, 1), User(2, "lisi", 19, 0), User(3, "wangwu",
20, 1)))
val df1: DataFrame = rdd.toDF()
df1.show()
// 直接创建 DataFrame
val df2 = spark.read.json("data/user/user.json")
df2.show()
// 创建临时表
df2.createOrReplaceTempView("t_user")
// 编写 SQL
lazy val sql = "SELECT id, name, age, gender FROM t_user ORDER BY age DESC"
// 执行 SQL
spark.sql(sql).show()
// ==================== 关闭连接 ====================
spark.stop
}

3.2.Dataset

def main(args: Array[String]): Unit = {
// ==================== 建立连接 ====================
// 初始化配置对象并设置运行模式与 AppName
val conf = new SparkConf().setMaster("local[*]").setAppName("DatasetDemo")
// 根据配置对象初始化 SparkSession 对象
val spark = SparkSession.builder().config(conf).getOrCreate()
// 日志级别
val sc = spark.sparkContext
sc.setLogLevel("ERROR")
// 在 IDEA 中开发 SparkSQL 如果遇到模型转换，需要导入隐式转换
import spark.implicits._
// ==================== 业务处理 ====================
// RDD 转 Dataset
val rdd = sc.makeRDD(List(User(1, "zhangsan", 18, 1), User(2, "lisi", 19, 0), User(3, "wangwu",
20, 1)))
val ds1: Dataset[User] = rdd.toDS()
ds1.show()
// 创建 DataFrame
val df: DataFrame = spark.read.json("data/user/user.json")
// 通过 DataFrame 使用 as[类型] 转换为 DataSet
val ds2: Dataset[User] = df.as[User]
ds2.show()
// 创建临时表
ds2.createOrReplaceTempView("t_user")
// 编写 SQL
lazy val sql = "SELECT id, name, age, gender FROM t_user ORDER BY age DESC"
// 执行 SQL
spark.sql(sql).show
// ==================== 关闭连接 ====================
spark.stop
}

4.DSL 领域特定语言

        DSL 为 Domain Specific Language 的缩写，翻译过来为领域特定语言。简单理解就是 Spark 独有的结构化数据操作语法。

        此处不做赘述。

5.自定义函数

5.1.UDF用户定义普通函数

案例： 

object UDFDemo {
case class Emp(empno: Int, ename: String, job: String, mgr: Int, hiredate: String, sal: Double,
comm: Double, deptno: Int)
def main(args: Array[String]): Unit = {
// ==================== 建立连接 ====================
// 初始化配置对象并设置运行模式与 AppName
val conf = new SparkConf().setMaster("local[*]").setAppName("UDFDemo")
// 根据配置对象初始化 SparkSession 对象
val spark = SparkSession.builder().config(conf).getOrCreate()
// 日志级别
val sc = spark.sparkContext
sc.setLogLevel("ERROR")
// 在 IDEA 中开发 SparkSQL 如果遇到模型转换，需要导入隐式转换
import spark.implicits._
// ==================== 业务处理 ====================
// 数据准备
val df: DataFrame = spark.read
.option("header", "true")
.option("sep", ",")
.option("inferSchema", "true")
.csv("data/scott/emp.csv")
val emp: Dataset[Emp] = df.as[Emp]
emp.createOrReplaceTempView("emp")
// 注册 UDF 函数
val prefix_name = spark.udf.register("prefix_name", (name: String) => {
"Hello: " + name
})
// 在 SQL 中使用
val sql =
"""
|SELECT ename, prefix_name(ename) AS new_name FROM emp
|""".stripMargin
spark.sql(sql).show(5)
// 在 DSL 中使用
emp.select('job, prefix_name('job).as("new_job")).show(5)
// ==================== 关闭连接 ====================
spark.stop
}
}

5.2.UDAF用户定义聚合函数

案例：

object UDAFDemo03_Spark3 {
case class Emp(empno: Int, ename: String, job: String, mgr: Int, hiredate: String, sal: Double,
comm: Double, deptno: Int)
def main(args: Array[String]): Unit = {
// ==================== 建立连接 ====================
// 初始化配置对象并设置运行模式与 AppName
val conf = new SparkConf().setMaster("local[*]").setAppName("UDAFDemo02")
// 根据配置对象初始化 SparkSession 对象
val spark = SparkSession.builder().config(conf).getOrCreate()
// 日志级别
val sc = spark.sparkContext
sc.setLogLevel("ERROR")
// 在 IDEA 中开发 SparkSQL 如果遇到模型转换，需要导入隐式转换
import spark.implicits._
// ==================== 业务处理 ====================
// 数据准备
val df: DataFrame = spark.read
.option("header", "true")
.option("sep", ",")
.option("inferSchema", "true")
.csv("data/scott/emp.csv")
val emp: Dataset[Emp] = df.as[Emp]
emp.createOrReplaceTempView("emp")
// 注册 UDAF 函数（强类型自定义 UDAF 在 Spark 3.0.0 中的使用方式）
val my_avg = spark.udf.register("my_avg", functions.udaf(MyAvg))
// 在 SQL 中使用
val sql =
"""
|SELECT my_avg(sal) AS avg_sal FROM emp
|""".stripMargin
spark.sql(sql).show()
// 在 DSL 中使用
emp.select(my_avg('sal).as("avg_sal")).show()
// ==================== 关闭连接 ====================
spark.stop
}
// 缓存区数据的结构 Buff(求和, 计数)
case class Buff(var sum: Double, var count: Long)
/**
* 自定义 UDAF 聚合函数：计算薪资的平均值
* IN：输入数据的类型
* BUFF：缓存区数据的类型
* OUT：返回值数据的类型
*/
object MyAvg extends Aggregator[Double, Buff, Double] {
// 初始化缓冲区 Buff(求和, 计数)
override def zero: Buff = Buff(0D, 0L)
// 根据输入的数据更新缓冲区的数据
override def reduce(b: Buff, in: Double): Buff = {
// 累加每次输入的数据
b.sum += in
// 计数器每次 +1
b.count += 1
// 返回缓冲区对象
b
}
// 合并缓冲区
override def merge(b1: Buff, b2: Buff): Buff = {
b1.sum += b2.sum
b1.count += b2.count
b1
}
// 计算最终结果
override def finish(b: Buff): Double = b.sum / b.count
// 缓冲区数据的编码处理
// Encoders.product 是进行 Scala 元组和 case 类转换的编码器
//override def bufferEncoder: Encoder[Buff] = Encoders.product
// 或者
override def bufferEncoder: Encoder[Buff] = Encoders.kryo(classOf[Buff])
// 输出数据的编码处理
override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

5.3.UDTF用户定义表创建函数

        先添加依赖：

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.12</artifactId>
<version>3.3.2</version>
</dependency>
<dependency>
<groupId>org.json</groupId>
<artifactId>json</artifactId>
<version>20220924</version>
</dependency>

     案例：

/**
* 数据：{"movie": [{"movie_name": "肖申克的救赎", "movie_type": "犯罪" }, {"movie_name": "肖申克的救赎",
"movie_type": "剧情" }]}
* 需求：从一行 JSON 格式数据中取出 movie_name 和 movie_type 两个 Key 及其对应的 Value。K-V 输出的格式为：
* movie_name movie_type
* 肖申克的救赎 犯罪
* 肖申克的救赎 剧情
*/
class MyUDTF extends GenericUDTF {
// 实例化 UDTF 对象，判断传入参数的长度以及数据类型
// 和 Hive 的自定义 UDTF 不一样的是，Spark 使用的是已经过时的 initialize(ObjectInspector[] argOIs)
override def initialize(argOIs: Array[ObjectInspector]): StructObjectInspector = {
// 获取入参
// 参数校验，判断传入参数的长度以及数据类型
if (argOIs.length != 1) throw new UDFArgumentLengthException("参数个数必须为 1")
if (ObjectInspector.Category.PRIMITIVE != argOIs(0).getCategory) {
/*
UDFArgumentTypeException(int argumentId, String message)
异常对象需要传入两个参数：
int argumentId：参数的位置，ObjectInspector 中的下标
String message：异常提示信息
*/
throw new UDFArgumentTypeException(0, "参数类型必须为 String")
}
// 自定义函数输出的字段和类型
// 创建输出字段名称的集合
val columNames = new util.ArrayList[String]
// 创建字段数据类型的集合
val columType = new util.ArrayList[ObjectInspector]
columNames.add("movie_name")
columType.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector)
columNames.add("movie_type")
columType.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector)
ObjectInspectorFactory.getStandardStructObjectInspector(columNames, columType)
}
// 处理数据
override def process(objects: Array[AnyRef]): Unit = {
val outline = new Array[String](2)
if (objects(0) != null) {
val jsonObject = new JSONObject(objects(0).toString)
val jsonArray: JSONArray = jsonObject.getJSONArray("movie")
var i = 0
while ( {
i < jsonArray.length
}) {
outline(0) = jsonArray.getJSONObject(i).getString("movie_name")
outline(1) = jsonArray.getJSONObject(i).getString("movie_type")
// 将处理好的数据通过 forward 方法将数据按行写出
forward(outline)
i += 1
}
}
}
override def close(): Unit = {}
}