33-SparkSql的介绍、DataFrame和DataSet

33-SparkSql的介绍、DataFrame和DataSet
十八、Spark SQL

18.1 简介

 18.1.1 Shark
1. 介绍
  - Shark 是基于 Spark 计算框架之上且兼容 Hive 语法的 SQL 执行引擎
  - 由于底层的计算采用了Spark ，性能比 MapReduce 的 Hive 普遍快2倍以上，当数据全部加载在内存的话，将快10倍以上，因此 Shark 可以作为交互式查询应用服务来使用
2. 特点
  - Shark 是完全兼容 Hive 的语法，表结构以及 UDF 函数等
  - 已有的Hive Sql 可以直接进行迁移至 Shark 上 Shark 底层依赖于 Hive 的解析器，查询优化器，
3. SparkSQL出现的原因
  - 但正是由于 shark 的整体设计架构对 Hive 的依赖性太强，难以支持其长远发展，比如不能和 Spark 的其他组件进行很好的集成，无法满足 Spark 的一站式解决大数据处理的需求。
18.1.2 发展
- Shark 是 SparkSQL 的前身, SparkSQL 产生的根本原因是其完全脱离了 Hive 的限制。
- SparkSQL 支持查询原生的 RDD 。 RDD 是 Spark 的核心概念，是 Spark 能够高效的处理大数据的各种场景的基础。
- 能够在 scala / java 中写 SQL 语句。支持简单的 SQL 语法检查，能够在 SQL 中写 Hive 语句访问Hive 数据，并将结果取回作为 RDD 使用
18.1.3 Spark on Hive和Hive on Spark
1. Spark on Hive ：
  - Hive只作为储存角色，Spark负责sql解析优化，执行。
2. Hive on Spark ：
  - Hive即作为存储又负责sql的解析优化，Spark负责执行
18.1.4 SparkSQL的特点
1. 能够将 SQL 查询与 Spark 程序无缝混合，允许您使用 SQL 或 DataFrame API 对结构化数据行查询；
2. 支持多种开发语言；
3. 支持多达上百种的外部数据源，包括 Hive，Avro，Parquet，ORC，JSON 和 JDBC 等；
4. 支持 HiveQL 语法以及 Hive SerDes 和 UDF，允许你访问现有的 Hive 仓库；
5. 支持标准的 JDBC 和 ODBC 连接；
6. 支持优化器，列式存储和代码生成等特性；
7. 支持扩展并能保证容错
18.2 DataSet和DataFrame
- DataSet和DataFrame引入的原因及两者之间的关系
  - 在SparkSql中要求被操作的数据必须是结构化的,所以引入了俩种数据类型,DataFrame和DataSet。DataFrame是spark1.3之后引入的分布式集合,DataSet是spark1.6之后引入的分布式集合。
  - 在spark2.0之后,DataFrame和DataSet的API统一了,DataFrame是DataSet的子集,DataSet是DataFrame的扩展
  - (type DataFrame = org.apache.spark.sql.Dataset[org.apache.spark.sql.Row])
18.2.1 DataFrame
1. 理解
  - 可以简单的把DataFrame理解成RDD+schema元信息
2. 特征
  - 在spark中，DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库的二维表格
  - DataFrame带有schema元信息，即DataFrame所表示的二维数据集的每一列都带有名称和类型
  - DataFrame可以从很多的数据源构建对象，如已存在的RDD、结构化文件、外部数据库、Hive表等
  - RDD可以把它的内部元素看成是一个Java对象，DataFrame内部是一个个Row对象，它表示一行一行的数据
3. RDD和DataFrame的区别
  - 如上图所示，左侧的RDD[Person]虽然以Person为类型参数，但Spark框架本身不了解Person类的内部结构
  - 右侧的DataFrame却提供了详细的结构信息，DataFrame多了数据的结构信息，即schema
4. 优点
  - 提高执行效率
  - 减少数据读取
  - 执行优化
  - Spark的DataFrame 是基于 RDD 的一种数据类型，具有比 RDD 节省空间和更高运算效率的优点，对于大数据量的运算，分布式计算能突破 pandas 的瓶颈，而 Spark 则是分布式计算的典型代表
5. 映射关系的转换方式
  - 通过样例类（case class）可以将数据进行结构化的映射转换，从而可以通过SQL语句来对数据进行查询等操作
  - 也可以通过结构化类型（StructType）将数据进行映射转换
6. DataFrame的创建方式
  - case class方式创建
```
/**
* 10,ACCOUNTING,NEWYORK
* 20,RESEARCH,DALLAS
* 30,SALES,CHICAGO
* 40,OPERATIONS,BOSTON
*/
object HelloSparkCaseClass {
    def main(args: Array[String]): Unit = {
        //创建对象
        val sparkSession: SparkSession =SparkSession.builder().master("local").appName("Hello02SparkSql").getOrCreate()
        //日志级别
        val sparkContext: SparkContext = sparkSession.sparkContext
        sparkContext.setLogLevel("ERROR")
        //读取数据
        val lines: RDD[String] =sparkContext.textFile("src/main/resources/dept.sql")
        val depts: RDD[Dept] = lines.map(_.split(",")).map(ele => newDept(ele(0).toInt, ele(1), ele(2)))
        //开始隐式转换
        import sparkSession.implicits._
        val dataFrame: DataFrame = depts.toDF()
        //打印信息-------DSL风格
        dataFrame.show()
        dataFrame.printSchema()
        println(dataFrame.count())
        dataFrame.columns.foreach(println)
        dataFrame.select("deptno", "dname").show()
        dataFrame.select("deptno", "dname").filter("deptno = 20").show()
        dataFrame.groupBy("dname").count().show()
        //打印信息-------SQL风格
        dataFrame.createOrReplaceTempView("t_dept")
        sparkSession.sql("select * from t_dept").show()
    }
}

case class Dept(deptno: Int, dname: String, loc: String)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
```
  - StructType方式创建
```
/**
* 10,ACCOUNTING,NEWYORK
* 20,RESEARCH,DALLAS
* 30,SALES,CHICAGO
* 40,OPERATIONS,BOSTON
*/
object HelloSparkStruct {
    def main(args: Array[String]): Unit = {
        //创建对象
        val sparkSession: SparkSession =SparkSession.builder().master("local").appName("Hello02SparkSql").getOrCreate()
        //日志级别
        val sparkContext: SparkContext = sparkSession.sparkContext
        sparkContext.setLogLevel("ERROR")
        //读取数据
        val lines: RDD[String] =sparkContext.textFile("src/main/resources/dept.sql")
        val depts = lines.map(_.split(",")).map(ele => Row(ele(0).toInt,ele(1), ele(2)))
        //创建DataFrame
        val dataFrame: DataFrame = sparkSession.createDataFrame(depts,structType)
        //打印数据
        dataFrame.show()
    }
    //创建表格的类型
    val structType: StructType = StructType(
        List(
            StructField("deptno", DataTypes.IntegerType),
            StructField("dname", DataTypes.StringType),
            StructField("loc", DataTypes.StringType)
        )
    )
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
```
18.2.2 DataSet
1. 理解
  - DataSet是分布式的数据集合，DataSet提供了强类型支持，在RDD的每行数据加了类型约束
  - Datset是在spark1.6中新添加的接口。它集中了RDD的优点（强类型和可以使用强大的lambda函数）以及使用了sparkSQL优化的执行引擎
2. DataSet和DataFrame的关系
  - DataFrame（在2.X之后）实际上是DataSet的一个特例，即对Dataset的元素为Row时起了一个别名
3. DataSet和RDD的关系
  - Dataset的底层封装的是RDD，当RDD的泛型是Row类型的时候，我们也可以称它为
    DataFrame
  - type DataFrame = Dataset[Row]
4. DataSet的创建实现
```
import org.apache.spark.sql.{Dataset, SparkSession}
/**
* {"deptno":10,"dname":"ACCOUNTING","loc":"NEWYORK"}
* {"deptno":20,"dname":"RESEARCH","loc":"DALLAS"}
* {"deptno":30,"dname":"SALES","loc":"CHICAGO"}
* {"deptno":40,"dname":"OPERATIONS","loc":"BOSTON"}
*/
object HelloDataSetJson {
    def main(args: Array[String]): Unit = {
        //创建SQL环境
        val sparkSession =SparkSession.builder().master("local").appName("Hello02DataFrameAvg").getOrCreate()
        import sparkSession.implicits._
        val dataSet: Dataset[DeptJson] =sparkSession.read.json("src/main/data/dept.json").as[DeptJson]
        //打印数据
        dataSet.show()
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
```
18.2.3 RDD, DataFrame ,DataSet 比较
1. 理解RDD, DataFrame ,DataSet
  - dataFrame在内存中映射为一张表,
  - RDD相当于表中一行数据,
  - dataset是具备RDD和dataFrame所有优点,强数据类型,保证编译时数据类型安全,符合面向对象编程,便于使用lamba函数
在spark2 版本中 dataFram源码已被移除,但是约定 DataFrame-DataSet[Row]
1. 主要区别
  - RDD可以知道每个元素具体类型,不知道元素具体属性
  - DataFrame数据类Row类型,数据列以及名称
  - DataSet 数据类型,字段名,字段类型
18.3 DSL数据操作

 18.3.1 Action操作
1. show
  - show（） 以表格的形式在输出中展示DataFrame中的数据，类似于 select * from spark_sql_test 的功能。
  - show 只显示前20条记录。
  - show(numRows: Int) 显示 numRows 条
  - show(truncate: Boolean) 是否最多只显示20个字符，默认为 true 。
  - show(numRows: Int, truncate: Boolean) 综合前面的显示记录条数，以及对过长字符串的显示格式。
2. collect
  - collect 方法会将 jdbcDF（DataFrame）中的所有数据都获取到，并返回一个 Array 对象
3. collectAsList
  - 获取所有数据到List
4. describe(cols: String*)
  - 获取指定字段的统计信息
5. first, head, take, takeAsList
  - 获取若干行记录
  - first 获取第一行记录
  - head 获取第一行记录， head(n: Int) 获取前n行记录
  - take(n: Int) 获取前n行数据
  - takeAsList(n: Int) 获取前n行数据，并以 List 的形式展现
18.3.2 查询
1. where(conditionExpr: String) ：
  - SQL语言中where关键字后的条件，可以用 and 和 or 。得到DataFrame类型的返回结果
2. filter ：根据字段进行筛选
  - 得到DataFrame类型的返回结果。和 where 使用条件相同
3. select ：获取指定字段值
  - 根据传入的 String 类型字段名，获取指定字段的值，以DataFrame类型返回
4. selectExpr ：可以对指定字段进行特殊处理
  - 可以直接对指定字段调用UDF函数，或者指定别名等。传入 String 类型参数，得到DataFrame对象。
5. col ：获取指定字段
  - 只能获取一个字段，返回对象为Column类型。
6. apply ：获取指定字段
  - 只能获取一个字段，返回对象为Column类型
7. drop ：去除指定字段，保留其他字段
  - 返回一个新的DataFrame对象，其中不包含去除的字段，一次只能去除一个字段
18.3.3 Limit
- limit 方法获取指定DataFrame的前n行记录，得到一个新的DataFrame对象
18.3.4 排序
1. orderBy 和 sort ：按指定字段排序，默认为升序
  - 按指定字段排序。加个 - 表示降序排序。 sort 和 orderBy 使用方法相同
  - jdbcDF.orderBy(- jdbcDF(“c4”)).show(false)
  - jdbcDF.orderBy(jdbcDF(“c4”).desc).show(false)
2. sortWithinPartitions
  - 和上面的 sort 方法功能类似，区别在于 sortWithinPartitions 方法返回的是按Partition排好序的DataFrame对象。
18.3.5 组函数
1. groupBy ：根据字段进行 group by 操作
  - groupBy 方法有两种调用方式，可以传入 String 类型的字段名，也可传入 Column 类型的对象。
2. cube 和 rollup ：group by的扩展
  - 功能类似于 SQL 中的 group by cube/rollup
3. GroupedData对象
  - 该方法得到的是 GroupedData 类型对象，在 GroupedData 的API中提供了 group by 之后的操作，
18.3.6 去重
1. distinct ：返回一个不包含重复记录的DataFrame
  - 返回当前DataFrame中不重复的Row记录。该方法和接下来的 dropDuplicates() 方法不传
    入指定字段时的结果相同。
2. dropDuplicates ：根据指定字段去重
  - 根据指定字段去重。类似于 select distinct a, b 操作
18.3.7 聚合
1. agg
  - 聚合操作调用的是 agg 方法，该方法有多种调用方式。一般与 groupBy 方法配合使用。
2. 示例
  - 以下示例其中最简单直观的一种用法，对 id 字段求最大值，对 c4 字段求和
```
jdbcDF.agg("id" -> "max", "c4" -> "sum")
1
```
18.3.8 Union
- unionAll 方法：对两个DataFrame进行组合 ,类似于 SQL 中的 UNION ALL 操作。
18.3.9 Join
1. 笛卡尔积
  - joinDF1.join(joinDF2)
2. using 一个字段形式
  - 下面这种join类似于 a join b using column1 的形式，需要两个DataFrame中有相同的一
    个列名，
  - joinDF1.join(joinDF2, “id”)
3. using 多个字段形式
  - 上面这种 using 一个字段的情况外，还可以 using 多个字段
18.3.10 Save
1. 作用
  - save可以将data数据保存到指定的区域
2. 使用
```
dataFrame.write.format("json").mode(SaveMode.Overwrite).save()
1
```
18.4 SparkSQL的数据源

数据处理的流程：数据源—>DataSet—>DSL操作数据

 18.4.1SparkSQL底层架构
1. Spark任务转化流程
  - 首先拿到 sql 后解析一批未被解决的逻辑计划，
  - 再经过分析得到分析后的逻辑计划，
  - 再经过一批优化规则转换成一批最佳优化的逻辑计划，
  - 再经过 SparkPlanner 的策略转化成一批物理计划，
  - 随后经过消费模型转换成一个个的 Spark 任务执行
2. SparkSQL执行流程
18.4.2 谓词下推

谓词下推是Catalyst优化器的两个重要优化方案之一
1. 理解
  - Predicate Pushdown简称谓词下推，简而言之，就是在不影响结果的情况下，尽量将过滤条件提前执行
2. 作用
  - 谓词下推后，过滤条件在map端执行，减少了map端的输出，降低了数据在集群上传输的量，节约了集群的资源，也提升了任务的性能
3. 图解
  
  [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-dU2YUN0h-1658574052102)(C:/Users/18446/AppData/Roaming/Typora/typora-user-images/image-20220723183614389.png)]
相关阅读:
如何根据性能需求进行场景设计？
vue 生命周期
 2.10 负载均衡原理- url hash 与 least_conn
ClickHouse Senior Course Ⅴ
HTML5:七天学会基础动画网页7
iOS小技能：日历的使用（案例：两个时间的比较、获取最近30天的数据）
idea运行springboot时报错Process finished with exit code 1
基于k-mean算法的医学图像分割matlab仿真
 Python自学教程5-字符串有哪些常用操作
 Idea中 css 、js 压缩插件会自动生成xxx.min.css、xxx.min.js文件
原文地址：https://blog.csdn.net/weixin_50627985/article/details/125951499

十八、Spark SQL

18.1 简介

18.1.1 Shark

18.1.2 发展

18.1.3 Spark on Hive和Hive on Spark

18.1.4 SparkSQL的特点

18.2 DataSet和DataFrame

18.2.1 DataFrame

18.2.2 DataSet

18.2.3 RDD, DataFrame ,DataSet 比较

18.3 DSL数据操作

18.3.1 Action操作

18.3.2 查询

18.3.3 Limit

18.3.4 排序

18.3.5 组函数

18.3.6 去重

18.3.7 聚合

18.3.8 Union

18.3.9 Join

18.3.10 Save

18.4 SparkSQL的数据源

18.4.1SparkSQL底层架构

18.4.2 谓词下推