Spark 系列（八）—— Spark SQL 之 DataFrame 和 Dataset
阿新 • • 發佈：2019-08-12
## 一、Spark SQL簡介
Spark SQL 是 Spark 中的一個子模組，主要用於操作結構化資料。它具有以下特點：
+ 能夠將 SQL 查詢與 Spark 程式無縫混合，允許您使用 SQL 或 DataFrame API 對結構化資料進行查詢；
+ 支援多種開發語言；
+ 支援多達上百種的外部資料來源，包括 Hive，Avro，Parquet，ORC，JSON 和 JDBC 等；
+ 支援 HiveQL 語法以及 Hive SerDes 和 UDF，允許你訪問現有的 Hive 倉庫；
+ 支援標準的 JDBC 和 ODBC 連線；
+ 支援優化器，列式儲存和程式碼生成等特性；
+ 支援擴充套件並能保證容錯。
## 二、DataFrame & DataSet
### 2.1 DataFrame
為了支援結構化資料的處理，Spark SQL 提供了新的資料結構 DataFrame。DataFrame 是一個由具名列組成的資料集。它在概念上等同於關係資料庫中的表或 R/Python 語言中的 `data frame`。 由於 Spark SQL 支援多種語言的開發，所以每種語言都定義了 `DataFrame` 的抽象，主要如下：
| 語言 | 主要抽象 |
| ------ | -------------------------------------------- |
| Scala | Dataset[T] & DataFrame (Dataset[Row] 的別名) |
| Java | Dataset[T] |
| Python | DataFrame |
| R | DataFrame |
### 2.2 DataFrame 對比 RDDs
DataFrame 和 RDDs 最主要的區別在於一個面向的是結構化資料，一個面向的是非結構化資料，它們內部的資料結構如下：
### 2.4 靜態型別與執行時型別安全
靜態型別 (Static-typing) 與執行時型別安全 (runtime type-safety) 主要表現如下:
在實際使用中，如果你用的是 Spark SQL 的查詢語句，則直到執行時你才會發現有語法錯誤，而如果你用的是 DataFrame 和 Dataset，則在編譯時就可以發現錯誤 (這節省了開發時間和整體代價)。DataFrame 和 Dataset 主要區別在於：
在 DataFrame 中，當你呼叫了 API 之外的函式，編譯器就會報錯，但如果你使用了一個不存在的欄位名字，編譯器依然無法發現。而 Dataset 的 API 都是用 Lambda 函式和 JVM 型別物件表示的，所有不匹配的型別引數在編譯時就會被發現。
以上這些最終都被解釋成關於型別安全圖譜，對應開發中的語法和分析錯誤。在圖譜中，Dataset 最嚴格，但對於開發者來說效率最高。
這裡一個可能的疑惑是 DataFrame 明明是有確定的 Scheme 結構 (即列名、列欄位型別都是已知的)，但是為什麼還是無法對列名進行推斷和錯誤判斷，這是因為 DataFrame 是 Untyped 的。
### 2.5 Untyped & Typed
在上面我們介紹過 DataFrame API 被標記為 `Untyped API`，而 DataSet API 被標記為 `Typed API`。DataFrame 的 `Untyped` 是相對於語言或 API 層面而言，它確實有明確的 Scheme 結構，即列名，列型別都是確定的，但這些資訊完全由 Spark 來維護，Spark 只會在執行時檢查這些型別和指定型別是否一致。這也就是為什麼在 Spark 2.0 之後，官方推薦把 DataFrame 看做是 `DatSet[Row]`，Row 是 Spark 中定義的一個 `trait`，其子類中封裝了列欄位的資訊。
相對而言，DataSet 是 `Typed` 的，即強型別。如下面程式碼，DataSet 的型別由 Case Class(Scala) 或者 Java Bean(Java) 來明確指定的，在這裡即每一行資料代表一個 `Person`，這些資訊由 JVM 來保證正確性，所以欄位名錯誤和型別錯誤在編譯的時候就會被 IDE 所發現。
```java
case class Person(name: String, age: Long)
val dataSet: Dataset[Person] = spark.read.json("people.json").as[Person]
```
## 三、DataFrame & DataSet & RDDs 總結
這裡對三者做一下簡單的總結：
+ RDDs 適合非結構化資料的處理，而 DataFrame & DataSet 更適合結構化資料和半結構化的處理；
+ DataFrame & DataSet 可以通過統一的 Structured API 進行訪問，而 RDDs 則更適合函數語言程式設計的場景；
+ 相比於 DataFrame 而言，DataSet 是強型別的 (Typed)，有著更為嚴格的靜態型別檢查；
+ DataSets、DataFrames、SQL 的底層都依賴了 RDDs API，並對外提供結構化的訪問介面。
## 四、Spark SQL的執行原理
DataFrame、DataSet 和 Spark SQL 的實際執行流程都是相同的：
1. 進行 DataFrame/Dataset/SQL 程式設計；
2. 如果是有效的程式碼，即程式碼沒有編譯錯誤，Spark 會將其轉換為一個邏輯計劃；
3. Spark 將此邏輯計劃轉換為物理計劃，同時進行程式碼優化；
4. Spark 然後在叢集上執行這個物理計劃 (基於 RDD 操作) 。
### 4.1 邏輯計劃(Logical Plan)
執行的第一個階段是將使用者程式碼轉換成一個邏輯計劃。它首先將使用者程式碼轉換成 `unresolved logical plan`(未解決的邏輯計劃)，之所以這個計劃是未解決的，是因為儘管您的程式碼在語法上是正確的，但是它引用的表或列可能不存在。 Spark 使用 `analyzer`(分析器) 基於 `catalog`(儲存的所有表和 `DataFrames` 的資訊) 進行解析。解析失敗則拒絕執行，解析成功則將結果傳給 `Catalyst` 優化器 (`Catalyst Optimizer`)，優化器是一組規則的集合，用於優化邏輯計劃，通過謂詞下推等方式進行優化，最終輸出優化後的邏輯執行計劃。
### 4.2 物理計劃(Physical Plan)
得到優化後的邏輯計劃後，Spark 就開始了物理計劃過程。 它通過生成不同的物理執行策略，並通過成本模型來比較它們，從而選擇一個最優的物理計劃在叢集上面執行的。物理規劃的輸出結果是一系列的 RDDs 和轉換關係 (transformations)。
### 4.3 執行
在選擇一個物理計劃後，Spark 執行其 RDDs 程式碼，並在執行時執行進一步的優化，生成本地 Java 位元組碼，最後將執行結果返回給使用者。
## 參考資料
1. Matei Zaharia, Bill Chambers . Spark: The Definitive Guide[M] . 2018-02
2. [Spark SQL, DataFrames and Datasets Guide](https://spark.apache.org/docs/latest/sql-programming-guide.html)
3. [且談 Apache Spark 的 API 三劍客：RDD、DataFrame 和 Dataset(譯文)](https://www.infoq.cn/article/three-apache-spark-apis-rdds-dataframes-and-datasets)
4. [A Tale of Three Apache Spark APIs: RDDs vs DataFrames and Datasets(原文)](https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html)
> **更多大資料系列文章可以參見 GitHub 開源專案**： [**大資料入門指南**](https://github.com/heibaiying/BigDat
## 二、DataFrame & DataSet
### 2.1 DataFrame
為了支援結構化資料的處理，Spark SQL 提供了新的資料結構 DataFrame。DataFrame 是一個由具名列組成的資料集。它在概念上等同於關係資料庫中的表或 R/Python 語言中的 `data frame`。 由於 Spark SQL 支援多種語言的開發，所以每種語言都定義了 `DataFrame` 的抽象，主要如下：
| 語言 | 主要抽象 |
| ------ | -------------------------------------------- |
| Scala | Dataset[T] & DataFrame (Dataset[Row] 的別名) |
| Java | Dataset[T] |
| Python | DataFrame |
| R | DataFrame |
### 2.2 DataFrame 對比 RDDs
DataFrame 和 RDDs 最主要的區別在於一個面向的是結構化資料，一個面向的是非結構化資料，它們內部的資料結構如下：
### 2.4 靜態型別與執行時型別安全
靜態型別 (Static-typing) 與執行時型別安全 (runtime type-safety) 主要表現如下:
在實際使用中，如果你用的是 Spark SQL 的查詢語句，則直到執行時你才會發現有語法錯誤，而如果你用的是 DataFrame 和 Dataset，則在編譯時就可以發現錯誤 (這節省了開發時間和整體代價)。DataFrame 和 Dataset 主要區別在於：
在 DataFrame 中，當你呼叫了 API 之外的函式，編譯器就會報錯，但如果你使用了一個不存在的欄位名字，編譯器依然無法發現。而 Dataset 的 API 都是用 Lambda 函式和 JVM 型別物件表示的，所有不匹配的型別引數在編譯時就會被發現。
以上這些最終都被解釋成關於型別安全圖譜，對應開發中的語法和分析錯誤。在圖譜中，Dataset 最嚴格，但對於開發者來說效率最高。
這裡一個可能的疑惑是 DataFrame 明明是有確定的 Scheme 結構 (即列名、列欄位型別都是已知的)，但是為什麼還是無法對列名進行推斷和錯誤判斷，這是因為 DataFrame 是 Untyped 的。
### 2.5 Untyped & Typed
在上面我們介紹過 DataFrame API 被標記為 `Untyped API`，而 DataSet API 被標記為 `Typed API`。DataFrame 的 `Untyped` 是相對於語言或 API 層面而言，它確實有明確的 Scheme 結構，即列名，列型別都是確定的，但這些資訊完全由 Spark 來維護，Spark 只會在執行時檢查這些型別和指定型別是否一致。這也就是為什麼在 Spark 2.0 之後，官方推薦把 DataFrame 看做是 `DatSet[Row]`，Row 是 Spark 中定義的一個 `trait`，其子類中封裝了列欄位的資訊。
相對而言，DataSet 是 `Typed` 的，即強型別。如下面程式碼，DataSet 的型別由 Case Class(Scala) 或者 Java Bean(Java) 來明確指定的，在這裡即每一行資料代表一個 `Person`，這些資訊由 JVM 來保證正確性，所以欄位名錯誤和型別錯誤在編譯的時候就會被 IDE 所發現。
```java
case class Person(name: String, age: Long)
val dataSet: Dataset[Person] = spark.read.json("people.json").as[Person]
```
## 三、DataFrame & DataSet & RDDs 總結
這裡對三者做一下簡單的總結：
+ RDDs 適合非結構化資料的處理，而 DataFrame & DataSet 更適合結構化資料和半結構化的處理；
+ DataFrame & DataSet 可以通過統一的 Structured API 進行訪問，而 RDDs 則更適合函數語言程式設計的場景；
+ 相比於 DataFrame 而言，DataSet 是強型別的 (Typed)，有著更為嚴格的靜態型別檢查；
+ DataSets、DataFrames、SQL 的底層都依賴了 RDDs API，並對外提供結構化的訪問介面。
## 四、Spark SQL的執行原理
DataFrame、DataSet 和 Spark SQL 的實際執行流程都是相同的：
1. 進行 DataFrame/Dataset/SQL 程式設計；
2. 如果是有效的程式碼，即程式碼沒有編譯錯誤，Spark 會將其轉換為一個邏輯計劃；
3. Spark 將此邏輯計劃轉換為物理計劃，同時進行程式碼優化；
4. Spark 然後在叢集上執行這個物理計劃 (基於 RDD 操作) 。
### 4.1 邏輯計劃(Logical Plan)
執行的第一個階段是將使用者程式碼轉換成一個邏輯計劃。它首先將使用者程式碼轉換成 `unresolved logical plan`(未解決的邏輯計劃)，之所以這個計劃是未解決的，是因為儘管您的程式碼在語法上是正確的，但是它引用的表或列可能不存在。 Spark 使用 `analyzer`(分析器) 基於 `catalog`(儲存的所有表和 `DataFrames` 的資訊) 進行解析。解析失敗則拒絕執行，解析成功則將結果傳給 `Catalyst` 優化器 (`Catalyst Optimizer`)，優化器是一組規則的集合，用於優化邏輯計劃，通過謂詞下推等方式進行優化，最終輸出優化後的邏輯執行計劃。
### 4.2 物理計劃(Physical Plan)
得到優化後的邏輯計劃後，Spark 就開始了物理計劃過程。 它通過生成不同的物理執行策略，並通過成本模型來比較它們，從而選擇一個最優的物理計劃在叢集上面執行的。物理規劃的輸出結果是一系列的 RDDs 和轉換關係 (transformations)。
### 4.3 執行
在選擇一個物理計劃後，Spark 執行其 RDDs 程式碼，並在執行時執行進一步的優化，生成本地 Java 位元組碼，最後將執行結果返回給使用者。
## 參考資料
1. Matei Zaharia, Bill Chambers . Spark: The Definitive Guide[M] . 2018-02
2. [Spark SQL, DataFrames and Datasets Guide](https://spark.apache.org/docs/latest/sql-programming-guide.html)
3. [且談 Apache Spark 的 API 三劍客：RDD、DataFrame 和 Dataset(譯文)](https://www.infoq.cn/article/three-apache-spark-apis-rdds-dataframes-and-datasets)
4. [A Tale of Three Apache Spark APIs: RDDs vs DataFrames and Datasets(原文)](https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html)
> **更多大資料系列文章可以參見 GitHub 開源專案**： [**大資料入門指南**](https://github.com/heibaiying/BigDat