概述

Spark SQL 是 Spark 用于結構化資料 ( structured data ) 處理的 Spark 模塊

Hive and SparkSQL

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

Hive 是早期唯一運行在 Hadoop 上的 SQL-on-Hadoop 工具

但 MapReduce 計算程序中大量的中間磁盤落地程序消耗了大量的 I / O，降低的運行效率，為了提高 SQL-on-Hadoop 的效率，大量的 SQL-on-Hadoop 工具開始產生

如 ：

• Drill
• Impala
• Shark

其中 Shark 是伯克利實驗室 Spark 生態環境的組件之一，是基于 Hive 所開發的工具，修改三個模塊，就運行在 Spark 引擎上

三個模塊 :

• 記憶體管理
• 物理計劃
• 執行

Shark 的出現，使得 SQL-on-Hadoop 的性能比 Hive 有了 10 - 100 倍的提高

最后"野心勃勃"，停止對 Shark 的開發，發展出兩個支線：SparkSQL 和 Hive on Spark

Spark SQL 為了簡化 RDD 的開發，提供了2個編程抽象 :

• DataFrame
• DataSet

特點

• 易整合 : 無縫的整合了 SQL 查詢和 Spark 編程
• 統一的資料訪問 : 使用相同的方式連接不同的資料源
• 兼容Hive : 使用相同的方式連接不同的資料源
• 標準資料連接 : 通過 JDBC 或者 ODBC 來連接

DataFrame

DataFrame 是一種以 RDD 為基礎的分布式資料集，類似于傳統資料庫中的二維表格

DataFrame 與 RDD 的區別 :

DataFrame : 有 schema 元資訊，即 DataFrame 所表示的二維表資料集的每一列都帶有名稱和型別，Spark SQL有更多的結構資訊，對 DataFrame 的資料源 和 在 DataFrame 之上的變換進行針對性的優化，可以提升運行效率

RDD : 不知資料元素的內部結構，Spark Core 只能在 stage 層面進行簡單、通用的流水線優化，

DataFrame 支持嵌套資料型別（struct、array 和 map）, DataFrame API提供了一套高層的關系操作，比函式式的 RDD API 更吊

DataFrame 和 RDD 的區別 :

• RDD[Person] : 以 Person 為型別引數，無類的內部結構

• DataFrame : 詳細的結構資訊，資料集中包含哪些列，每列的名稱和型別各是什么

DataFrame : 為資料提供了 Schema 的視圖，當資料庫中的一張表來對待

DataFrame 也是懶執行的，但性能上比 RDD 要高

主要原因：優化的執行計劃，即查詢計劃通過 Spark catalyst optimiser 進行優化

例子 :

user.join(events, users("id") === events("uids"))
	.filter(events("date") > "2021-11-11")

左圖 : 構造了兩個 DataFrame ，將它們 join 之后又做了一次 filter 操作 , 執行該執行計劃，執行效率不高

因為 join 是一個代價較大的操作，也可能會產生一個較大的資料集

右圖 : 將 filter下推到 join下方，先對 DataFrame 進行過濾，再 join 過濾后的較小的結果集，就可縮短執行時間

Spark SQL 的查詢優化器的邏輯查詢計劃優化就是一個利用基于關系代數的等價變換，將高成本的操作替換為低成本操作的程序

性能圖 :

DataSet

DataSet 是分布式資料集合

DataSet 是 Spark 1.6 中添加的一個新抽象，是 DataFrame 的一個擴展，提供了 RDD 的優勢（強型別， lambda 函式）以及 Spark SQL 優化執行引擎的優點

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、Person 這些的型別一樣，所有的表結構資訊都用 Row 來表示，獲取資料時需要指定順序

核心編程

Spark Core 中，如果要執行應用程式，先構建背景關系環境物件 SparkContext ，Spark SQL 為對 Spark Core 的一種封裝，不僅在模型上封裝，還在背景關系環境物件也封裝了

以前 , SparkSQL有兩種SQL查詢起始點：

• SQLContext : Spark 自己提供的 SQ L查詢
• HiveContext : 連接 Hive 的查詢

SparkSession 是 Spark 最新的 SQL 查詢起始點，實質上是 SQLContext 和 HiveContext 的組合，所以在 SQLContex 和HiveContext 上可用的 API 在 SparkSession 上也可使用

SparkSession 內部封裝了 SparkContext，所以計算實際上是由 sparkContext 完成的

當使用 spark-shell 的時 , spark 會自動創建 SparkSession 物件 , 就和以前自動獲取 sc 來表示 SparkContext 物件一樣

DataFrame

DataFrame API 允許使用 DataFrame 而不用去注冊臨時表 或 生成 SQL 運算式

DataFrame API 既有 transformation 操作也有 action 操作

創建DataFrame

SparkSession 是創建 DataFrame 和執行 SQL 的入口

創建DataFrame有三種方式 :

• 通過 Spark 的資料源進行創建
• 從一個存在的 RDD 進行轉換
• Hive Table 進行查詢回傳

從Spark資料源進行創建

查看 Spark 支持創建檔案的資料源格式

spark.read.

在 spark 的 bin/data 目錄（windows環境）中創建 user.json 檔案

{
	"username":"zhangsan",
	"age":20
}

讀取 json 檔案創建 DataFrame

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

從記憶體中獲取資料，spark 可以知道資料型別具體是什么，如 : 數字，默認作為 Int 處理
從檔案中讀取的數字，不確定是什么型別，用 bigint 接收，可以和 Long 型別轉換，但不能和 Int 進行轉換

展示結果 :

show

從RDD進行轉換

從Hive Table進行查詢回傳

SQL語法

SQL 語法風格 : 查詢資料時使用 SQL 陳述句來查詢，這種風格的查詢必須要有臨時視圖或者全域視圖來輔助

讀取 JSON 檔案創建 DataFrame

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

對 DataFrame 創建一個臨時表(視圖)

df.createOrReplaceTempView("people")

通過 SQL 陳述句實作查詢全表

val sqlDF = spark.sql("SELECT * FROM people")

結果展示

sqlDF.show

普通臨時表是 Session 范圍內的
全域臨時表是應用范圍內有效，使用全域臨時表時需要全路徑訪問，如：global_temp.people

對于 DataFrame 創建一個全域表

df.createGlobalTempView("people")

通過 SQL 陳述句實作查詢全表

spark.sql("SELECT * FROM global_temp.people").show()

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

DSL語法

DataFrame 提供一個特定領域語言(domain-specific language, DSL)去管理結構化的資料，

可以在 Scala, Java, Python 和 R 中使用 DSL，使用 DSL 語法風格不用創建臨時視圖

創建一個DataFrame

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

查看 DataFrame 的 Schema 資訊

df.printSchema

只查看 “username” 列資料，

df.select("username").show()

查看 “username” 列資料以及 “age+1” 資料

涉及到運算時 , 每列都必須使用 $ , 或 用引號運算式：單引號 + 欄位名

df.select($"username", $"age" + 1).show()

df.select('username, ' age + 1).show

df.select('username, ' age + 1 as "newAge").show

查看 “age” 大于 “30” 的資料

df.filter($"age" > 30).show

按照 “age” 分組，查看資料條數

df.groupBy("age").count.show

RDD 轉 DataFrame

在 IDEA 中開發程式時，如果需要 RDD 與 DF 或者 DS 之間互相操作，那么需要引入 import spark.implicits._ , spark 不是 Scala 中的包名

先創建 SparkSession 物件再匯入 ，再創建的 sparkSession 物件的變數名稱

Scala 只支持 val 修飾的物件的引入 ，所以 spark 物件不能使用 var 宣告

spark-shell 中無需匯入，自動完成此操作

val idRDD = sc.textFile("data/id.txt")

idRDD.toDF("id").show

將 RDD 轉換為 DataFrame

case class User(name:String, age:Int)

sc.makeRDD(List(("cpucode",30), ("cpu",40)))
	.map(t => User(t._1, t._2)).toDF.show

DataFrame 轉 RDD

DataFrame 其實就是對 RDD 的封裝，所以可以直接獲取內部的 RDD

val df = sc.makeRDD(List(("cpucode", 20), ("cpu", 10)))
	.map(t => User(t._1, t._2)).toDF

val rdd = df.rdd

val arrar = rdd.collect

得到的 RDD 存盤型別為 Row

array(0)

array(0)(0)

array(0).getAs[String]("name")

DataSet

DataSet 是具有強型別的資料集合，需要提供對應的型別資訊

創建 DataSet

樣例類序列創建 DataSet

case class Person(name: String, age: Long)

val caseClassDS = Seq(Person("cpucode", 22)).toDS()

caseClassDS.show

基本型別的序列創建 DataSet

val ds = Seq(1, 2, 3, 4, 5).toDS

ds.show

RDD 轉 DataSet

SparkSQL 能夠自動將包含有 case 類的 RDD 轉換成 DataSet

case 類定義了 table 的結構，case 類屬性通過反射變成了表的列名，Case 類可以包含 Seq 或 Array 等復雜的結構

case class User(name:String, age:Int)

sc.makeRDD(List(("cpucode", 22), ("cpu", 33), ("code", 44)))
	.map(t => User(t._1, t._2)).toDS

DataSet 轉 RDD

DataSet 其實也是對 RDD 的封裝，所以可以直接獲取內部的 RDD

case class User(name: String, age: Int)

sc.makeRDD(List(("cpucode", 20), ("code", 11)))
	.map(t => User(t._1, t._2)).toDS

val rdd = res11.rdd

rdd.collect

DataFrame 和 DataSet轉換

DataFrame 其實是 DataSet 的特例，所以它們之間是可以互相轉換的

DataFrame 轉 DataSet

case class User(name: String, age: Int)

val df = sc.makeRDD(List("cpucode", 22), ("code", 33)).toDF("name", "age")

val ds = df.as[User]

DataSet 轉 DataFrame

val df = ds.toDF

RDD、DataFrame、DataSet三者的關系

SparkSQL 兩個新的抽象 :

• DataFrame
• DataSet

從版本的產生上來看：

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

三者的共性

RDD、DataFrame、DataSet 全都是 Spark 平臺下的分布式彈性資料集，為處理超大型資料提供便利

三者都有惰性機制，在進行創建、轉換，如 map 方法時，不會立即執行，只有在遇到 Action 如 foreach 時，三者才會開始遍歷運算

三者有許多共同的函式，如 filter，排序等

在對 DataFrame 和 Dataset 進行操作許多操作都需要這個包: import spark.implicits._（在創建好 SparkSession 物件后盡量直接匯入）

三者都會根據 Spark 的記憶體情況自動快取運算 ，這樣即使資料量很大，也不用擔心會記憶體溢位

三者都有 partition 的概念

DataFrame 和 DataSet 均可使用模式匹配獲取各個欄位的值和型別

三者的共性

RDD

• RDD 一般和 spark mllib 同時使用
• RDD不支持 sparksql 操作

DataFrame

• 與 RDD 和 Dataset 不同，DataFrame 每一行的型別固定為 Row，每一列的值沒法直接訪問，只有通過決議才能獲取各個欄位的值

• DataFrame 與 DataSet 一般不與 spark mllib 同時使用

• DataFrame 與 DataSet 均支持 SparkSQL 的操作，如 : select，groupby 之類，還能注冊臨時表 / 視窗，進行 sql 陳述句操作

• DataFrame 與 DataSet 支持一些保存方式，如 : 保存成 csv，可以帶上表頭，這樣每一列的欄位名

DataSet

• Dataset 和 DataFrame 擁有完全相同的成員函式，區別只是每一行的資料型別不同， DataFrame 是 DataSet 的一個特例 type DataFrame = Dataset[Row]

• DataFrame ( Dataset[Row] ) , 每一行的型別是 Row，不決議，每一行有哪些欄位什么型別不知道，只能用 getAS 方法 或 共性中模式匹配拿出特定欄位

• Dataset ，每一行是什么型別是不一定的，在自定義 case class 后可以獲得每一行的資訊

三者的互相轉換

IDEA開發SparkSQL

添加依賴

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

代碼實作

object SparkSql_Demo{
	def main(args: Array[String]): Unit = {
		//創建背景關系環境配置物件
    	val conf: SparkConf = new SparkConf()
    		.setMaster("local[*]").setAppName("SparkSql_Demo")
		
		//創建SparkSession物件
		val spark:SparkSession = SparkSession.builder().config(conf).getOrCreate()

	    //RDD => DataFrame => DataSet 轉換需要引入隱式轉換規則，否則無法轉換
	    //spark 不是包名，是背景關系環境物件名
		import spark.implicits._
	
	
		//讀取json檔案 創建DataFrame  {"username": "lisi","age": 18}
		val df: DataFrame = spark.read.json("input/test.json")
		//df.show()

		//SQL風格語法
		df.createOrReplaceTempView("user")
		//spark.sql("select avg(age) from user").show

		//DSL風格語法
		//df.select("username", "age").show()

		//***** RDD => DataFrame => DataSet *****
		//RDD
		val rdd1: RDD[(Int, String, Int)] = spark.sparkContext
			.makeRDD(List(1, "cpucode", 30), (2, "cpu", 22), (3, "code", 33))

		//DataFrame
		val df1: DataFrame = rdd1.toDF("id", "name", "age")
		//df1.show()
		
		//DateSet
		val ds1: Dataset[User] = df1.as[User]
	 	//ds1.show()


		//***** DataSet => DataFrame= > RDD *****
		//DataFrame
		val df2: DataFrame = ds1.toDF()
		
    	//RDD  回傳的RDD型別為Row，里面提供的getXXX方法可以獲取欄位值，類似jdbc處理結果集，但是索引從0開始
		val rdd2: RDD[Row] = df2.rdd
		
    	//rdd2.foreach(a=>println(a.getString(1)))
    
		//***** RDD => DataSet *****
		rdd1.map{
			case (id, name, age) => User(id, name, age)
		}.toDS()
		
		//*****DataSet=>=>RDD*****
		ds1.rdd

	    //釋放資源
	    spark.stop()
	}
}

case class User(id:Int, name:String, age:Int)

用戶自定義函式

通過 spark.udf 添加自定義函式，實作自定義功能

UDF

創建 DataFrame

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

注冊 UDF

spark.udf.register("addName", (x: String) => "Name:" + x)

創建臨時表

df.createOrReplaceTempView("people")

應用 UDF

spark.sql("select addName(name), age from people").show()

UDAF

強型別的 Dataset 和 弱型別的 DataFrame 都提供了相關的聚合函式

如 : count()，countDistinct()，avg()，max()，min()

用戶可以設定自己的自定義聚合函式，通過繼承 UserDefinedAggregateFunction 來實作用戶自定義弱型別聚合函式

從 Spark3.0 版本后，不推薦使用弱型別聚合 ，統一采用強型別聚合函式 Aggregator

需求：計算平均工資

實作方式 - RDD

val conf: SparkConf = new SparkConf().setAppName("app").setMaster("local[*]")
val sc: SparkContext = new SparkContext(conf)
val res: (Int, Int) = sc.makeRDD(List("cpucode", 22), ("lisi", 30)).map{
	case (name, age) => {
		(age, 1)
	}
}.reduce {
	(t1, t2) => {
		(t1._1 + t2._1, t1._2 + t2._2)
	}
}

println(res._1 / res._2)

// 關閉連接
sc.stop()

實作方式 - 累加器

class MyAC extends AccumulatorV2[Int, Int]{
	var sum:Int = 0


}

實作方式 - UDAF - 弱型別

實作方式 - UDAF - 強型別

Spark3.0 版本可以采用強型別的 Aggregate 方式代替 UserDefinedAggregateFunction

資料的加載和保存

通用的加載和保存方式

SparkSQL 提供了通用的保存資料和資料加載的方式

通用指的是使用相同的API，根據不同的引數讀取 和 保存不同格式的資料，SparkSQL默認讀取和保存的檔案格式為 parquet

加載資料

spark.read.load 是加載資料的通用方法

spark.read.

讀取不同格式的資料，可以對不同的資料格式進行設定

spark.read.format("...")[.option("...")].load("...")

• format("…") ：指定加載的資料型別，包括"csv"、“jdbc”、“json”、“orc”、“parquet"和"textFile”

在檔案上進行查詢: 檔案格式.`檔案路徑`

spark.sql("select * from json.`/opt/module/data/user.json`").show

保存資料

Scala/Java	Any Language	Meaning
SaveMode.ErrorIfExists(default)	“error”(default)	如果檔案已經存在則拋出例外
SaveMode.Append	“append”	如果檔案已經存在則追加
SaveMode.Overwrite	“overwrite”	如果檔案已經存在則覆寫
SaveMode.Ignore	“ignore”	如果檔案已經存在則忽略

df.write.mode("append").json("/opt/module/data/output")

Parquet

Spark SQL 的默認資料源為 Parquet 格式，Parquet 是一種能夠有效存盤嵌套資料的列式存盤格式

資料源為 Parquet 檔案時，Spark SQL 可以方便的執行所有的操作，不需要使用 format ，修改配置項 spark.sql.sources.default ，可修改默認資料源格式

JSON

Spark SQL 能夠自動推測 JSON 資料集的結構，并將它加載為一個 Dataset[Row]. 可以通過SparkSession.read.json() 去加載 JSON 檔案

Spark讀取的JSON檔案不是傳統的JSON檔案，每一行都應該是一個JSON串

CSV

MySQL

Hive

專案實戰