8 Spark_SQL 开荒

刺骨的言语ヽ痛彻心扉 2022-12-05 09:55 110阅读 0赞

![在这里插入图片描述][20200909163334116.png_pic_center]  
![在这里插入图片描述][2020090916333980.png_pic_center]

### 文章目录 ###

*  一. 概述
 *   *   *   *  Shark
             *  Shark
             *  Shark重新设计Spark SQL
             *  停止Shark
         *  1.2 SparkSQL特点
         *  1.3 DataFrame
         *  1.4 DataSet
 *  二. 新的起点
 *   *   *  2.2 DataFrame
         *   *  2.2.1 创建DataFrame
             *  2.2.2 SQL语法
             *  2.2.3 DSL语法
             *  2.2.4 RDD转换为DataFrame
             *  2.2.5 DataFrame转换为RDD
     *  2.3 DataSet
     *   *   *  2.3.2 RDD转换为DataSet
             *  2.3.3 DataSet转换为RDD
     *  2.4 DataFrame和DataSet转换
     *  2.5 RDD DS DF
     *   *   *  2.5.1 RDD DS DF 共同点
             *  2.5.2 RDD DS DF区别

# 一. 概述 #

#### Shark ####

SparkSQL的前身是Shark，给熟悉RDBMS但又不理解MapReduce的技术人员提供快速上手的工具

Hive是早期唯一运行在Hadoop上的SQL-on-Hadoop工具。但是MapReduce计算过程中大量的中间磁盘落地过程消耗了大量的I/O，降低的运行效率，为了提高SQL-on-Hadoop的效率，大量的SQL-on-Hadoop工具开始产生，其中表现较为突出的是：

1.  Drill
2.  Impala
3.  Shark

#### Shark ####

Shark是伯克利实验室Spark生态环境的组件之一，是基于Hive所开发的工具

![在这里插入图片描述][20200909163705858.png_pic_center]  
修改了内存管理、物理计划、执行三个模块 ， 使Shark能运行在Spark引擎上

Shark的出现，使得SQL-on-Hadoop的性能比Hive有了10-100倍的提高。(官方说的，数据量大起来还是MR)

#### Shark重新设计Spark SQL ####

但是，随着Spark的发展，对于野心勃勃的Spark团队来说，Shark对于Hive的太多依赖（如采用Hive的语法解析器、查询优化器等等），制约了Spark的One Stack Rule Them All的既定方针，制约了Spark各个组件的相互集成，所以提出了SparkSQL项目

SparkSQL抛弃原有的Sharkdiamagnetic，汲取一点Shark的优点，（内存列存储、Hive兼容性），完全重新开发

1.  数据兼容方面：但兼容Hive，还可以从RDD、parquet文件、JSON文件中获取数据，未来版本甚至支持获取RDBMS数据以及cassandra等NOSQL数据。
2.  性能优化：内存列存储、byte-code generation 、引进Cost Model对查询进行动态评估、获取最佳物理计划等等；
3.  组件方面 : 重新设计了SQL语法解析器、分析器、优化器

![在这里插入图片描述][20200909164645125.png_pic_center]

#### 停止Shark ####

2014年6月1日Shark项目和SparkSQL项目的主持人Reynold Xin宣布：停止对Shark的开发，团队将所有资源放SparkSQL项目上，至此，Shark的发展画上了句话，但也因此发展出两个支线：SparkSQL和Hive on Spark  
![在这里插入图片描述][20200909164916786.png_pic_center]  
其中SparkSQL作为Spark生态的一员继续发展，而不再受限于Hive，只是兼容Hive；而Hive on Spark是一个Hive的发展计划，该计划将Spark作为Hive的底层引擎之一，也就是说，Hive将不再受限于一个引擎，可以采用Map-Reduce、Tez、Spark等引擎。

### 1.2 SparkSQL特点 ###

1.  易整合  
    ![在这里插入图片描述][20200909165046945.png_pic_center]
2.  统一的数据访问

相同的方式链接不同的数据源

![在这里插入图片描述][20200909165114131.png_pic_center]  
3) 兼容Hive

在已有的仓库上直接运行 SQL 或者 HiveQL

![在这里插入图片描述][20200909165203732.png_pic_center]

1.  标准数据连接  
    JDBC & ODBC

![在这里插入图片描述][20200909165227407.png_pic_center]

### 1.3 DataFrame ###

DataFrame : 以RDD为基础的分布式数据集，类似传统数据库的二维表格

比RDD多了schema元信息，也就是DataFrame所表示的二维表数据集的每一列都带有名称和类型

从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标  
![在这里插入图片描述][20200909165858649.png_pic_center]  
DataFrame是为数据提供了Schema的视图。可以把它当做数据库中的一张表来对待

DataFrame也是懒执行的，但性能上比RDD要高，主要原因：优化的执行计划，即查询计划通过Spark catalyst optimiser进行优化。比如下面一个例子:  
![在这里插入图片描述][20200909170003117.png_pic_center]  
![在这里插入图片描述][20200909170006917.png_pic_center]  
Spark SQL的源于RDD 但是性能方面又比RDD要高，原因 SparkSQL会自己做优化的RDD  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 1]

### 1.4 DataSet ###

DataSet是分布式数据集合。DataSet是Spark 1.6中添加的一个新抽象，  
都是对RDD的封装，提供了RDD的优势(强类型，lambda函数能力)以及SparkSQL优化执行引擎的优点  
DataSet也可以进行功能性的转换(操作map，flatMap,filter等)

*  DataSet是DataFrame API的一个扩展，是SparkSQL最新的数据抽象
 *  API友好，类型安全检查与DataFrame的查询优化
 *  用样例类来对DataSet中定义的数据的结构信息，样例类中每个属性的名称直接映射到DataSet中的字段名称；
 *  DataSet是强类型的：DataSet\[Car\] DataSet\[Person\]
 *  DataFrame 是 DataSet的特例，DataFrame = DataSet\[Row\] , 所以可以通过as 方法将DataFrame 转成DataSet。  
    Row为一个类型， 和 Car Persron一样，搜友的表结构信息都可以用Row来表示，获取数据时需要指定顺序

# 二. 新的起点 #

![在这里插入图片描述][20200922001911100.png_pic_center]  
Spark context 为Spark Core的起点  
Spark session 是Spark SQL 的起点

SparkContext，Spark SQL其实可以理解为对Spark Core的一种封装，不仅仅在模型上进行了封装，上下文环境对象也进行了封装。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 2]

### 2.2 DataFrame ###

Spark SQL的DataFrame API 允许我们使用 DataFrame 而不用必须去注册临时表或者生成 SQL 表达式。DataFrame API 既有 transformation操作也有action操作。

#### 2.2.1 创建DataFrame ####

SparkSession是创建DataFrame和执行SQL的入口，创建DataFrame有三种方式：1 数据源 2 RDD转换 3 Hive Table

1.  从Spark数据源进行创建

//查看Spark支持创建文件的数据源格式
    scala> spark.read.
    csv   format   jdbc   json   load   option   options   orc   parquet   schema   table   text   textFile

val df = spark.read.json("data/user.json")

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 3]  
这里接受完数据就可以直接看到数据类型，  
如果数据直接从 内存 中读取， 数字默认 Int  
如果数据从磁盘中读取，不确定是什么类型，所以用bigint接受 ， 可以和Long类型转换 ，但不能转Int

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 4]

#### 2.2.2 SQL语法 ####

写SQL必须要临时视图 or 全局视图来辅助

1.  读取Json文件创建DataFrame

val df = spark.read.json("data/user.json")
    df: org.apache.spark.sql.DataFrame = [age: bigint， username: string]

1.  对DataFrame创建一个临时表

df.createOrReplaceTemView("people")

1.  实现SQL

scala> val sqlDF = spark.sql("SELECT * FROM people")
    sqlDF: org.apache.spark.sql.DataFrame = [age: bigint， name: string]

**普通表是Session范围内的，全局临时表是应用范围内有效的**

1.  创建全局临时视图

df.createGlobalTempView("people")

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 5]

#### 2.2.3 DSL语法 ####

弥补df创建view的麻烦

特定领域语言(domain-specific language, DSL)去管理结构化的数据。

用对象的方式去操作数据

1.  查看DataFrame 的Schema信息

scala> df.printSchema
    root
     |-- age: Long (nullable = true)
     |-- username: string (nullable = true)

1.  查看username列数据

1.  查询 age + 1 的数据

\*\*涉及运算的时候，每列都必须使用$ or ‘ \*\*

scala> df.select($"username",$"age" + 1).show
    scala> df.select('username, 'age + 1).show()
    scala> df.select('username, 'age + 1 as "newage").show()
    
    
    +--------+---------+
    |username|(age + 1)|
    +--------+---------+
    |zhangsan|        21|
    |     lisi|        31|
    |  wangwu|         41|
    +--------+---------+

1.  查看“age”大于 30 的数据

scala> df.filter($"age">30).show
    +---+---------+
    |age| username|
    +---+---------+
    | 40|    wangwu|
    +---+---------+

1.  查看age 大于 30 的数据

df.filter('age > 30).show

1.  按照“年龄”来进行分组，查看数据的条数

df.groupBy('age).count.show

#### 2.2.4 RDD转换为DataFrame ####

需要引入

import spark.implicits._

spark对象不能使用var声明，因为Scala只支持val修饰的对象的引入。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 7]  
**一般是通过样例类将RDD转换成为DataFrame**

case class User(id:Int,name:String,age:Int)
    
    val userRDD = rdd.map( t => User(t._1,t._2,t._3))
    
    val df = userRDD.toDF

#### 2.2.5 DataFrame转换为RDD ####

DataFrame其实就是对RDD的封装，所以可以直接获取内部的RDD

scala> val df = sc.makeRDD(List(("zhangsan",30), ("lisi",40))).map(t=>User(t._1, t._2)).toDF
    df: org.apache.spark.sql.DataFrame = [name: string, age: int]
    
    scala> val rdd = df.rdd   //df 有对rdd的包装
    rdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[46] at rdd at <console>:25
    
    scala> val array = rdd.collect
    array: Array[org.apache.spark.sql.Row] = Array([zhangsan,30], [lisi,40])  //把结构包装成了row（行）对象

**注意：此时得到的RDD存储类型为Row**

## 2.3 DataSet ##

DataSet是具有强类型的数据集合，需要提供对应的类型信息。

DafaFrame只有结构 ，没有类型

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 8]

1.  使用样例类创建DataSet

scala> val caseClassDS = Seq(Person("zhangsan",2)).toDS()
    
    caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: Long]

scala> val ds = Seq(1,2,3,4,5).toDS
    ds: org.apache.spark.sql.Dataset[Int] = [value: int]

#### 2.3.2 RDD转换为DataSet ####

需要给RDD添加一系列类型  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 10]  
SparkSQL能够自动将包含case类的RDD转成DataSet  
case定义了table的结构，case类属性通过反射成为表的列名  
case类可以包含Seq 或者Array等复制的机构

case User(name:string,age:Int)
    
    sc.makeRDD(list(("zhangsan",30),("lisi",40)).map(t => User(t._1,t._2)).toDS

#### 2.3.3 DataSet转换为RDD ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 11]

DataSet其实也是对RDD的封装，所以可以直接获取内部的RDD

scala> case class User(name:String, age:Int)
    defined class User
    
    scala> sc.makeRDD(List(("zhangsan",30), ("lisi",49))).map(t=>User(t._1, t._2)).toDS
    res11: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
    
    scala> val rdd = res11.rdd
    rdd: org.apache.spark.rdd.RDD[User] = MapPartitionsRDD[51] at rdd at <console>:25
    
    scala> rdd.collect
    res12: Array[User] = Array(User(zhangsan,30), User(lisi,49))

## 2.4 DataFrame和DataSet转换 ##

DataFrame其实是DataSet的特例，所以它们之间是可以互相转换的

1.  DataFrame 转换成 DataSet

scala> case class User(name:String, age:Int)
    defined class User
    
    scala> val df = sc.makeRDD(List(("zhangsan",30), ("lisi",49))).toDF("name","age")
    df: org.apache.spark.sql.DataFrame = [name: string, age: int]
    
    scala> val ds = df.as[User]   //定义 row ----> User
    ds: org.apache.spark.sql.Dataset[User] = [name: string, age: int]

1.  DataSet 转换成 DataFrame

scala> val ds = df.as[User]
    ds: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
    
    scala> val df = ds.toDF
    df: org.apache.spark.sql.DataFrame = [name: string, age: int]

## 2.5 RDD DS DF ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 12]

Spark1.0 => RDD 
    	Spark1.3 => DataFrame
    	Spark1.6 => Dataset

#### 2.5.1 RDD DS DF 共同点 ####

 RDD、DataFrame、DataSet全都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利;  
 三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action如foreach时，三者才会开始遍历运算;  
 三者有许多共同的函数，如filter，排序等;  
 在对DataFrame和Dataset进行操作许多操作都需要这个包:import spark.implicits.\_（在创建好SparkSession对象后尽量直接导入）  
 三者都会根据 Spark 的内存情况自动缓存运算，这样即使数据量很大，也不用担心会内存溢出  
 三者都有partition的概念  
 DataFrame和DataSet均可使用模式匹配获取各个字段的值和类型

#### 2.5.2 RDD DS DF区别 ####

1.  DataFrame

DF 就是 DS的 一个特例 type DataFrame = Dataset\[Row\]  
每一行的数据类型都为Row 不解析 每一行的数据类型不知

DF 每一行类型固定为Row，每一列的值没法直接访问 ，只能先解析

[20200909163334116.png_pic_center]: /images/20221123/0ce16a7c6bf44e53a4e3f383123cd1ff.png
[2020090916333980.png_pic_center]: /images/20221123/1f14e5480eaf40d0913a4918da490869.png
[20200909163705858.png_pic_center]: /images/20221123/6a7068d4e3f34643931eae9fb08e92aa.png
[20200909164645125.png_pic_center]: /images/20221123/8014582b2322432c8b84d6b7725eea94.png
[20200909164916786.png_pic_center]: /images/20221123/710d21710b854b9ab9192a8b47add96c.png
[20200909165046945.png_pic_center]: /images/20221123/49fa340a6cfe47e08d094c8d3baf11c3.png
[20200909165114131.png_pic_center]: /images/20221123/9fbb4d62458445aa9405d9884bdb2084.png
[20200909165203732.png_pic_center]: /images/20221123/28b1bc06e6b740f092b05e1a48f7860a.png
[20200909165227407.png_pic_center]: /images/20221123/2cc548a07fc049af8495ecfaf6b0b08c.png
[20200909165858649.png_pic_center]: /images/20221123/08a32bf89a5f4f5c9f986f1c0fc1de9c.png
[20200909170003117.png_pic_center]: /images/20221123/38048a37ce4e48f0b6a73fc87f27abac.png
[20200909170006917.png_pic_center]: /images/20221123/6279d6b3815041de8913fc4aaa8a253c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/c60c16beb5ad475584d588c4c280014c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/defecbac5e2b4faab9a5f315fd639ba3.png
[20200922001911100.png_pic_center]: /images/20221123/37c252a13c1e40439d8d18e7a7f611c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/da2d778d176947eeb932e0a2c3516c86.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/df672d89ee074d98a594ae723ec7f032.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/f7de24a9a4c6473fa873b77729ded6a9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/4df65df3892f4b8cb52f426f1389dd2a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/837398a29e534217a8faa92d12ce6af1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221123/9b2c0ff988d240478bf53039951547a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221123/cb97712647b843689e399f1c30de07ad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221123/b64d28339c64488393021550d30871d0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221123/9fbe385c71c643b18a5b3fb8ca88a54c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20221123/ced6b35ac5584b15be10b2dc324312cc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 12]: /images/20221123/e00c828dd7dc4afc9fc22f86f1c49e22.png