Spark 系列（八）—— Spark SQL 之 DataFrame 和 Dataset

一、Spark SQL簡(jiǎn)介

Spark SQL 是 Spark 中的一個(gè)子模塊，主要用于操作結(jié)構(gòu)化數(shù)據(jù)。它具有以下特點(diǎn)：

• 能夠?qū)?SQL 查詢與 Spark 程序無(wú)縫混合，允許您使用 SQL 或 DataFrame API 對(duì)結(jié)構(gòu)化數(shù)據(jù)進(jìn)行查詢；
• 支持多種開(kāi)發(fā)語(yǔ)言；
• 支持多達(dá)上百種的外部數(shù)據(jù)源，包括 Hive，Avro，Parquet，ORC，JSON 和 JDBC 等；
• 支持 HiveQL 語(yǔ)法以及 Hive SerDes 和 UDF，允許你訪問(wèn)現(xiàn)有的 Hive 倉(cāng)庫(kù)；
• 支持標(biāo)準(zhǔn)的 JDBC 和 ODBC 連接；
• 支持優(yōu)化器，列式存儲(chǔ)和代碼生成等特性；
• 支持?jǐn)U展并能保證容錯(cuò)。

二、DataFrame & DataSet

2.1 DataFrame

為了支持結(jié)構(gòu)化數(shù)據(jù)的處理，Spark SQL 提供了新的數(shù)據(jù)結(jié)構(gòu) DataFrame。DataFrame 是一個(gè)由具名列組成的數(shù)據(jù)集。它在概念上等同于關(guān)系數(shù)據(jù)庫(kù)中的表或 R/Python 語(yǔ)言中的 data frame。 由于 Spark SQL 支持多種語(yǔ)言的開(kāi)發(fā)，所以每種語(yǔ)言都定義了 DataFrame 的抽象，主要如下：

2.2 DataFrame 對(duì)比 RDDs

DataFrame 和 RDDs 最主要的區(qū)別在于一個(gè)面向的是結(jié)構(gòu)化數(shù)據(jù)，一個(gè)面向的是非結(jié)構(gòu)化數(shù)據(jù)，它們內(nèi)部的數(shù)據(jù)結(jié)構(gòu)如下：

DataFrame 內(nèi)部的有明確 Scheme 結(jié)構(gòu)，即列名、列字段類(lèi)型都是已知的，這帶來(lái)的好處是可以減少數(shù)據(jù)讀取以及更好地優(yōu)化執(zhí)行計(jì)劃，從而保證查詢效率。

DataFrame 和 RDDs 應(yīng)該如何選擇？

• 如果你想使用函數(shù)式編程而不是 DataFrame API，則使用 RDDs；
• 如果你的數(shù)據(jù)是非結(jié)構(gòu)化的 (比如流媒體或者字符流)，則使用 RDDs，
• 如果你的數(shù)據(jù)是結(jié)構(gòu)化的 (如 RDBMS 中的數(shù)據(jù)) 或者半結(jié)構(gòu)化的 (如日志)，出于性能上的考慮，應(yīng)優(yōu)先使用 DataFrame。

2.3 DataSet

Dataset 也是分布式的數(shù)據(jù)集合，在 Spark 1.6 版本被引入，它集成了 RDD 和 DataFrame 的優(yōu)點(diǎn)，具備強(qiáng)類(lèi)型的特點(diǎn)，同時(shí)支持 Lambda 函數(shù)，但只能在 Scala 和 Java 語(yǔ)言中使用。在 Spark 2.0 后，為了方便開(kāi)發(fā)者，Spark 將 DataFrame 和 Dataset 的 API 融合到一起，提供了結(jié)構(gòu)化的 API(Structured API)，即用戶可以通過(guò)一套標(biāo)準(zhǔn)的 API 就能完成對(duì)兩者的操作。

這里注意一下：DataFrame 被標(biāo)記為 Untyped API，而 DataSet 被標(biāo)記為 Typed API，后文會(huì)對(duì)兩者做出解釋。

2.4 靜態(tài)類(lèi)型與運(yùn)行時(shí)類(lèi)型安全

靜態(tài)類(lèi)型 (Static-typing) 與運(yùn)行時(shí)類(lèi)型安全 (runtime type-safety) 主要表現(xiàn)如下:

在實(shí)際使用中，如果你用的是 Spark SQL 的查詢語(yǔ)句，則直到運(yùn)行時(shí)你才會(huì)發(fā)現(xiàn)有語(yǔ)法錯(cuò)誤，而如果你用的是 DataFrame 和 Dataset，則在編譯時(shí)就可以發(fā)現(xiàn)錯(cuò)誤 (這節(jié)省了開(kāi)發(fā)時(shí)間和整體代價(jià))。DataFrame 和 Dataset 主要區(qū)別在于：

在 DataFrame 中，當(dāng)你調(diào)用了 API 之外的函數(shù)，編譯器就會(huì)報(bào)錯(cuò)，但如果你使用了一個(gè)不存在的字段名字，編譯器依然無(wú)法發(fā)現(xiàn)。而 Dataset 的 API 都是用 Lambda 函數(shù)和 JVM 類(lèi)型對(duì)象表示的，所有不匹配的類(lèi)型參數(shù)在編譯時(shí)就會(huì)被發(fā)現(xiàn)。

以上這些最終都被解釋成關(guān)于類(lèi)型安全圖譜，對(duì)應(yīng)開(kāi)發(fā)中的語(yǔ)法和分析錯(cuò)誤。在圖譜中，Dataset 最嚴(yán)格，但對(duì)于開(kāi)發(fā)者來(lái)說(shuō)效率最高。

<div align="center"> <img width="600px" src="https://raw.githubusercontent.com/heibaiying/BigData-Notes/master/pictures/spark-運(yùn)行安全.png"/> </div>
上面的描述可能并沒(méi)有那么直觀，下面的給出一個(gè) IDEA 中代碼編譯的示例：

這里一個(gè)可能的疑惑是 DataFrame 明明是有確定的 Scheme 結(jié)構(gòu) (即列名、列字段類(lèi)型都是已知的)，但是為什么還是無(wú)法對(duì)列名進(jìn)行推斷和錯(cuò)誤判斷，這是因?yàn)?DataFrame 是 Untyped 的。

2.5 Untyped & Typed

在上面我們介紹過(guò) DataFrame API 被標(biāo)記為 Untyped API，而 DataSet API 被標(biāo)記為 Typed API。DataFrame 的 Untyped 是相對(duì)于語(yǔ)言或 API 層面而言，它確實(shí)有明確的 Scheme 結(jié)構(gòu)，即列名，列類(lèi)型都是確定的，但這些信息完全由 Spark 來(lái)維護(hù)，Spark 只會(huì)在運(yùn)行時(shí)檢查這些類(lèi)型和指定類(lèi)型是否一致。這也就是為什么在 Spark 2.0 之后，官方推薦把 DataFrame 看做是 DatSet[Row]，Row 是 Spark 中定義的一個(gè) trait，其子類(lèi)中封裝了列字段的信息。

相對(duì)而言，DataSet 是 Typed 的，即強(qiáng)類(lèi)型。如下面代碼，DataSet 的類(lèi)型由 Case Class(Scala) 或者 Java Bean(Java) 來(lái)明確指定的，在這里即每一行數(shù)據(jù)代表一個(gè) Person，這些信息由 JVM 來(lái)保證正確性，所以字段名錯(cuò)誤和類(lèi)型錯(cuò)誤在編譯的時(shí)候就會(huì)被 IDE 所發(fā)現(xiàn)。

case class Person(name: String, age: Long)
val dataSet: Dataset[Person] = spark.read.json("people.json").as[Person]

三、DataFrame & DataSet & RDDs 總結(jié)

這里對(duì)三者做一下簡(jiǎn)單的總結(jié)：

• RDDs 適合非結(jié)構(gòu)化數(shù)據(jù)的處理，而 DataFrame & DataSet 更適合結(jié)構(gòu)化數(shù)據(jù)和半結(jié)構(gòu)化的處理；
• DataFrame & DataSet 可以通過(guò)統(tǒng)一的 Structured API 進(jìn)行訪問(wèn)，而 RDDs 則更適合函數(shù)式編程的場(chǎng)景；
• 相比于 DataFrame 而言，DataSet 是強(qiáng)類(lèi)型的 (Typed)，有著更為嚴(yán)格的靜態(tài)類(lèi)型檢查；
• DataSets、DataFrames、SQL 的底層都依賴了 RDDs API，并對(duì)外提供結(jié)構(gòu)化的訪問(wèn)接口。

<div align="center"> <img width="600px" src="https://raw.githubusercontent.com/heibaiying/BigData-Notes/master/pictures/spark-structure-api.png"/> </div>

四、Spark SQL的運(yùn)行原理

DataFrame、DataSet 和 Spark SQL 的實(shí)際執(zhí)行流程都是相同的：

1. 進(jìn)行 DataFrame/Dataset/SQL 編程；
2. 如果是有效的代碼，即代碼沒(méi)有編譯錯(cuò)誤，Spark 會(huì)將其轉(zhuǎn)換為一個(gè)邏輯計(jì)劃；
3. Spark 將此邏輯計(jì)劃轉(zhuǎn)換為物理計(jì)劃，同時(shí)進(jìn)行代碼優(yōu)化；
4. Spark 然后在集群上執(zhí)行這個(gè)物理計(jì)劃 (基于 RDD 操作) 。

4.1 邏輯計(jì)劃(Logical Plan)

執(zhí)行的第一個(gè)階段是將用戶代碼轉(zhuǎn)換成一個(gè)邏輯計(jì)劃。它首先將用戶代碼轉(zhuǎn)換成 unresolved logical plan(未解決的邏輯計(jì)劃)，之所以這個(gè)計(jì)劃是未解決的，是因?yàn)楸M管您的代碼在語(yǔ)法上是正確的，但是它引用的表或列可能不存在。 Spark 使用 analyzer(分析器) 基于 catalog(存儲(chǔ)的所有表和 DataFrames 的信息) 進(jìn)行解析。解析失敗則拒絕執(zhí)行，解析成功則將結(jié)果傳給 Catalyst 優(yōu)化器 (Catalyst Optimizer)，優(yōu)化器是一組規(guī)則的集合，用于優(yōu)化邏輯計(jì)劃，通過(guò)謂詞下推等方式進(jìn)行優(yōu)化，最終輸出優(yōu)化后的邏輯執(zhí)行計(jì)劃。

4.2 物理計(jì)劃(Physical Plan)

得到優(yōu)化后的邏輯計(jì)劃后，Spark 就開(kāi)始了物理計(jì)劃過(guò)程。 它通過(guò)生成不同的物理執(zhí)行策略，并通過(guò)成本模型來(lái)比較它們，從而選擇一個(gè)最優(yōu)的物理計(jì)劃在集群上面執(zhí)行的。物理規(guī)劃的輸出結(jié)果是一系列的 RDDs 和轉(zhuǎn)換關(guān)系 (transformations)。

4.3 執(zhí)行

在選擇一個(gè)物理計(jì)劃后，Spark 運(yùn)行其 RDDs 代碼，并在運(yùn)行時(shí)執(zhí)行進(jìn)一步的優(yōu)化，生成本地 Java 字節(jié)碼，最后將運(yùn)行結(jié)果返回給用戶。

參考資料

1. Matei Zaharia, Bill Chambers . Spark: The Definitive Guide[M] . 2018-02
2. Spark SQL, DataFrames and Datasets Guide
3. 且談 Apache Spark 的 API 三劍客：RDD、DataFrame 和 Dataset(譯文)
4. A Tale of Three Apache Spark APIs: RDDs vs DataFrames and Datasets(原文)

更多大數(shù)據(jù)系列文章可以參見(jiàn) GitHub 開(kāi)源項(xiàng)目： 大數(shù)據(jù)入門(mén)指南