Spark SQL

一、简介

Spark SQL 是 Spark 用于结构化数据(structured data)处理的 Spark 模块。

1.特点:

➢ 数据兼容方面 SparkSQL 不但兼容 Hive，还可以从 RDD、parquet 文件、JSON 文件中获取数据，未来版本甚至支持获取 RDBMS 数据以及 cassandra 等 NOSQL 数据；

➢ 性能优化方面除了采取 In-Memory Columnar Storage、byte-code generation 等优化技术外、将会引进 Cost Model 对查询进行动态评估、获取最佳物理计划等等；

➢ 组件扩展方面无论是 SQL 的语法解析器、分析器还是优化器都可以重新定义，进行扩展。

2.优点

SparkSQL 可以简化 RDD 的开发，提高开发效率，且执行效率非常快，所以实际工作中，基本上采用的就是 SparkSQL。Spark SQL 为了简化 RDD 的开发，提高开发效率，提供了 2 个编程抽象，类似 Spark Core 中的 RDD

DataFrame、DataSet.

3.RDD、DataFrame、DataSet 三者的关系

在 SparkSQL 中 Spark 为我们提供了两个新的抽象，分别是 DataFrame 和 DataSet

3.1.首先从版本的产生上来看：

➢ Spark1.0 => RDD

➢ Spark1.3 => DataFrame

➢ Spark1.6 => Dataset

如果同样的数据都给到这三个数据结构，他们分别计算之后，都会给出相同的结果。不同是的他们的执行效率和执行方式.

在后期的 Spark 版本中，DataSet 有可能会逐步取代 RDD和 DataFrame 成为唯一的 API 接口。

3.2.三者的共性

➢ RDD、DataFrame、DataSet 全都是 spark 平台下的分布式弹性数据集，为处理超大型数据提供便利;

➢ 三者都有惰性机制，在进行创建、转换，如 map 方法时，不会立即执行，只有在遇到Action 如 foreach 时，三者才会开始遍历运算;

➢ 三者有许多共同的函数，如 filter，排序等;

➢ 在对 DataFrame 和 Dataset 进行操作许多操作都需要这个包:import spark.implicits._（在创建好 SparkSession 对象后尽量直接导入）

➢ 三者都会根据 Spark 的内存情况自动缓存运算，这样即使数据量很大，也不用担心会内存溢出

➢ 三者都有 partition 的概念

➢ DataFrame 和 DataSet 均可使用模式匹配获取各个字段的值和类型

3.3.三者的区别

3.3.1.RDD

➢ RDD 一般和 spark mllib 同时使用

➢ RDD 不支持 spark sql 操作

3.3.2.DataFrame

➢ 与 RDD 和 Dataset 不同，DataFrame 每一行的类型固定为 Row，每一列的值没法直接访问，只有通过解析才能获取各个字段的值

➢ DataFrame 与 DataSet 一般不与 spark mllib 同时使用

➢ DataFrame 与 DataSet 均支持 SparkSQL 的操作，比如 select，groupby 之类，还能注册临时表/视窗，进行 sql 语句操作

➢ DataFrame 与 DataSet 支持一些特别方便的保存方式，比如保存成 csv，可以带上表头，这样每一列的字段名一目了然

3.3.3.DataSet

➢ Dataset 和 DataFrame 拥有完全相同的成员函数，区别只是每一行的数据类型不同。DataFrame 其实就是 DataSet 的一个特例 type DataFrame = Dataset[Row]

➢ DataFrame 也可以叫 Dataset[Row],每一行的类型是 Row，不解析，每一行究竟有哪些字段，各个字段又是什么类型都无从得知，只能用上面提到的 getAS 方法或者共性中的第七条提到的模式匹配拿出特定字段。而 Dataset 中，每一行是什么类型是不一定的，在自定义了 case class 之后可以很自由的获得每一行的信息

二、DataFrame

DataFrame 是 Spark SQL 中表示数据集的一种方式，它是以命名列组织的分布式数据集合，并支持多种数据源。可以通过 spark.read 方法读取不同的数据源，如 CSV、JDBC 等，来创建 DataFrame。

1.创建DataFrame

在 Spark SQL 中 SparkSession 是创建 DataFrame 和执行 SQL 的入口，

创建 DataFrame有三种方式：

通过 Spark 的数据源进行创建.
从一个存在的 RDD 进行转换.
还可以从 Hive Table 进行查询返回.

1.1.通过 Spark 的数据源进行创建

1.1.1.查看 Spark 支持创建文件的数据源格式

scala> spark.read.csv format jdbc json load option options orc parquet schema 
table text textFile

sparkdatauserjson__105">1.1.2.在 spark 的 data 目录中创建 user.json 文件

[{"username": "张三","age": 29},{"username": "里斯","age": 30},{"username": "王武","age": 37},{"username": "赵六","age": 41},{"username": "陈七","age": 52}
]

1.1.3.读取 json 文件创建 DataFrame

scala> val df=spark.read.option("multiLine",true).json("../data/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, username: string]

**注意事项:**如果json文件格式化了需要指定读取多行为真,option("multiLine",true),也可以不格式化json文件,直接将内容压平成为一个平面文件

1.1.4.结果展示

scala> df.show
+---+--------+
|age|username|
+---+--------+
| 29|    张三|
| 30|    里斯|
| 37|    王武|
| 41|    赵六|
| 52|    陈七|
+---+--------+

**注意：**如果从内存中获取数据，spark 可以知道数据类型具体是什么。如果是数字，默认作为 Int 处理；但是从文件中读取的数字，不能确定是什么类型，所以用 bigint 接收，可以和Long 类型转换，但是和 Int 不能进行转换

1.2.从一个存在的 RDD 进行转换

在后续章节中讨论

1.3.从 Hive Table 进行查询返回

在后续章节中讨论

2.SQL语法

SQL 语法风格是指我们查询数据的时候使用 SQL 语句来查询，这种风格的查询必须要有临时视图或者全局视图来辅助

2.1.1.对 DataFrame 创建一个临时表

scala> df.createOrReplaceTempView("people")

**注意:**此处是承接的前面读取的那个user.json文件,继续操作的

2.1.2.通过 SQL 语句实现查询全表

scala> val sqlDF=spark.sql("select * from people")
sqlDF: org.apache.spark.sql.DataFrame = [age: bigint, username: string]

2.1.3.结果展示

scala> sqlDF.show
+---+--------+
|age|username|
+---+--------+
| 29|    张三|
| 30|    里斯|
| 37|    王武|
| 41|    赵六|
| 52|    陈七|
+---+--------+

注意：普通临时表是 Session 范围内的，如果想应用范围内有效，可以使用全局临时表。使用全局临时表时需要全路径访问，如：global_temp.people

2.1.4.对于 DataFrame 创建一个全局表

scala> df.createGlobalTempView("people")

2.1.5.通过 SQL 语句实现查询全表

scala> spark.sql("SELECT * FROM global_temp.people").show()
或
scala> spark.newSession().sql("SELECT * FROM global_temp.people").show()

3.DSL 语法

DataFrame 提供一个特定领域语言(domain-specific language, DSL)去管理结构化的数据。可以在 Scala, Java, Python 和 R 中使用 DSL，使用 DSL 语法风格不必去创建临时视图了

3.1.1创建一个 DataFrame

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

3.1.2.查看 DataFrame 的 Schema 信息

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

3.1.3.只查看"username"列数据

scala> df.select("username").show
+--------+
|username|
+--------+
|    张三|
|    里斯|
|    王武|
|    赵六|
|    陈七|
+--------+

3.1.4.查看"username"列数据以及"age+1"数据

scala> df.select($"username",$"age"+1).show
+--------+---------+
|username|(age + 1)|
+--------+---------+
|    张三|       30|
|    里斯|       31|
|    王武|       38|
|    赵六|       42|
|    陈七|       53|
+--------+---------+
或者
scala> df.select('username,'age+1).show
+--------+---------+
|username|(age + 1)|
+--------+---------+
|    张三|       30|
|    里斯|       31|
|    王武|       38|
|    赵六|       42|
|    陈七|       53|
+--------+---------+

**注意:**涉及到运算的时候, 每列都必须使用$, 或者采用引号表达式：单引号+字段名

3.1.5.查看"age"大于"30"的数据

scala> df.filter($"age">30).show
+---+--------+
|age|username|
+---+--------+
| 37|    王武|
| 41|    赵六|
| 52|    陈七|
+---+--------+

3.1.6.按照"age"分组，查看数据条数

scala> df.groupBy("age").count.show
+---+-----+                                                                     
|age|count|
+---+-----+
| 29|    1|
| 52|    1|
| 41|    1|
| 37|    1|
| 30|    1|
+---+-----+

4.RDD 转换为 DataFrame

在 IDEA 中开发程序时，如果需要 RDD 与 DF 或者 DS 之间互相操作，那么需要引入import spark.implicits._

这里的 spark 不是 Scala 中的包名，而是创建的 sparkSession 对象的变量名称，所以必须先创建 SparkSession 对象再导入。这里的 spark 对象不能使用 var 声明，因为 Scala 只支持val 修饰的对象的引入。

sparkshell__316">4.1.1.spark-shell 中无需导入，自动完成此操作。

scala> val idRdd=sc.textFile("../data/id.txt")
scala> idRdd.toDF("id").show
+---+
| id|
+---+
| id|
|  1|
|  2|
|  3|
|  4|
|  5|
|  6|
+---+

4.1.2.实际开发中，一般通过样例类将 RDD 转换为 DataFrame

scala> case class User(name:String, age:Int)
defined class Userscala> sc.makeRDD(List(("张三",40), ("里斯",40))).map(t=>User(t._1, t._2)).toDF.show
+----+---+
|name|age|
+----+---+
|张三| 40|
|里斯| 40|
+----+---+

5.DataFrame转换为RDD

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

scala> val df=sc.makeRDD(List(("张三",40), ("里斯",40))).map(t=>User(t._1, t._2)).toDF
df: org.apache.spark.sql.DataFrame = [name: string, age: int]scala> val rdd=df.rdd
rdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[117] at rdd at <console>:26scala> val array=rdd.collect
array: Array[org.apache.spark.sql.Row] = Array([张三,40], [里斯,40])scala> array(0)
res50: org.apache.spark.sql.Row = [张三,40]scala> array(1)
res51: org.apache.spark.sql.Row = [里斯,40]

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

三、Dataset

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

1.创建DataSet

1.1.1使用样例类序列创建 DataSet


scala> case class Person(name: String, age: Long)
defined class Personscala> val caseClassDS = Seq(Person("张三",23),Person("里斯",43)).toDS()
caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]scala> caseClassDS.show
+----+---+
|name|age|
+----+---+
|张三| 23|
|里斯| 43|
+----+---+

1.1.2.使用基本类型的序列创建 DataSet

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

2.RDD转换为DataSet

SparkSQL 能够自动将包含有 case 类的 RDD 转换成 DataSet，case 类定义了 table 的结构，case 类属性通过反射变成了表的列名。Case 类可以包含诸如 Seq 或者 Array 等复杂的结构

scala>  case class User(name:String, age:Int)
defined class Userscala> sc.makeRDD(List(("赵六",30), ("陈七",49))).map(t=>User(t._1, t._2)).toDS.show
+----+---+
|name|age|
+----+---+
|赵六| 30|
|陈七| 49|
+----+---+

3.DataSet转换为RDD

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

scala>  case class User(name:String, age:Int)
defined class Userscala> val ds=sc.makeRDD(List(("昭武",40), ("武则天",69))).map(t=>User(t._1, t._2)).toDS
ds: org.apache.spark.sql.Dataset[User] = [name: string, age: int]scala> val rdd=ds.rdd
rdd: org.apache.spark.rdd.RDD[User] = MapPartitionsRDD[134] at rdd at <console>:26scala> rdd.collect
res59: Array[User] = Array(User(昭武,40), User(武则天,69))

4.DataFrame和DataSet转换

scala> case class User(name:String, age:Int)
defined class Userscala>val df = sc.makeRDD(List(("奥本",31), ("里斯本",59))).toDF("name","age")
df: org.apache.spark.sql.DataFrame = [name: string, age: int]//DataFrame 转换为 DataSet
scala>val ds=df.as[User]
res60: org.apache.spark.sql.Dataset[User] = [name: string, age: int]//DataSet 转换为 DataFrame
scala>ds.toDF
res64: org.apache.spark.sql.DataFrame = [name: string, age: int]