Spark SQL模組，主要就是處理跟SQL解析相關的一些內容，說得更通俗點就是怎麼把一個SQL語句解析成Dataframe或者說RDD的任務。以Spark 2.4.3為例，Spark SQL這個大模組分為三個子模組，如下圖所示

其中Catalyst可以說是Spark內部專門用來解析SQL的一個框架，在Hive中類似的框架是Calcite（將SQL解析成MapReduce任務）。Catalyst將SQL解析任務分成好幾個階段，這個在對應的論文中講述得比較清楚，本系列很多內容也會參考論文，有興趣閱讀原論文的可以到這裡看：Spark SQL: Relational Data Processing in Spark。

而Core模組其實就是Spark SQL主要解析的流程，當然這個過程中會去呼叫Catalyst的一些內容。這模組裡面比較常用的類包括SparkSession，DataSet等。

至於hive模組，這個不用說，肯定跟hive有關的。這個模組在本系列基本不會涉及到，就不多介紹了。

值得一提的是，論文發表的時候還是在Spark1.x階段，那個時候SQL解析成詞法樹用的是scala寫的一個解析工具，到2.x階段改為使用antlr4來做這部分工作（這應該算是最大的改變）。至於為什麼要改，我猜是出於可讀性和易用性方面的考慮，當然這個僅是個人猜測。

另外，這一系列會簡單介紹一條SQL語句的處理流程，基於spark 2.4.3（sql這個模組在spark2.1後變化不大）。這一篇先從整體介紹Spark SQL出現的背景及解決問題，Dataframe API以及Catalyst的流程大概是怎麼樣，後面分階段細說Catalyst的流程。

Spark SQL出現的背景及解決的問題

在最早的時候，大規模處理資料的技術是MapReduce，但這種框架執行效率太慢，進行一些關係型處理（如join）需要編寫大量程式碼。後來hive這種框架可以讓使用者輸入sql語句，自動進行優化並執行。

但在大型系統中，任然有兩個主要問題，一個是ETL操作需要對接多個數據源。另一個是使用者需要執行復雜分析，比如機器學習和圖計算等。但傳統的關係型處理系統中較難實現。

Spark SQL提供了兩個子模組來解決這個問題，DataFrame API和Catalyst。

相比於RDD，Dataframe api提供了更加豐富的關係型api，並且能和RDD相互轉換，後面Spark機器學習方面的工作重心，也從以RDD為基礎的mllib轉移到以Dataframe為基礎的Spark ML（雖然Dataframe底層也是RDD）。

另一個就是Catalyst，通過它可以輕鬆為諸如機器學習之類的域新增資料來源（比如json或通過case class自定義的型別），優化規則和資料型別。

通過這兩個模組，Spark SQL主要實現以下目標：

• 提供方便易用好的API，包括讀取外部資料來源，以及關係資料處理（用過的都知道）
• 使用已建立的DBMS技術提供高效能。
• 輕鬆支援新資料來源，包括半結構化資料和外部資料庫（比如MYSQL）。
• 圖計算和機器學習方面的拓展

那下面就介紹Dataframe和Catalyst的流程，當然主要討論的還是Catalyst。

統一API Dataframe

先來看看論文裡面提供的一張圖：

這張圖可以說明很多，首先Spark的Dataframe API底層也是基於Spark的RDD。但與RDD不同的在於，Dataframe會持有schema（這個實在不好翻譯，可以理解為資料的結構吧），以及可以執行各種各樣的關係型操作，比如Select，Filter，Join，Groupby等。從操作上來說，和pandas的Dataframe有點像（連名字都是一樣的）。

同時因為是基於RDD的，所以很多RDD的特性Dataframe都能夠享受到，比如說分散式計算中一致性，可靠性方面的保證，以及可以通過cache快取資料，提高計算效能啊等等。

同時圖中頁展示了Dataframe可以通過JDBC連結外部資料庫，通過控制檯操作（spark-shell），或者使用者程式。說白了，就是Dataframe可以通過RDD轉換而來，也可以通過外部資料表生成。

對了，這裡順便說一句，很多初次接觸Spark SQL的童鞋可能會對Dataset和Dataframe這兩個東西感到疑惑，在1.x時代它們確實有些差別，不過在spark2.x的時候，這兩個API已經統一了。所以基本上Dataset和Dataframe可以看成是等價的東西。

最後還是結合程式碼做一下實際的展示吧，如下展示生成一個RDD，並且根據這個RDD生成對應的Dataframe，從中可以看出RDD和Dataframe的區別：

//生成RDD
scala> val data = sc.parallelize(Array((1,2),(3,4)))
data: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> data.foreach(println)
(1,2)
(3,4)

scala> val df = data.toDF("fir","sec")
df: org.apache.spark.sql.DataFrame = [fir: int, sec: int]

scala> df.show()
+---+---+
|fir|sec|
+---+---+
|  1|  2|
|  3|  4|
+---+---+

//跟RDD相比，多了schema
scala> df.printSchema()
root
 |-- fir: integer (nullable = false)
 |-- sec: integer (nullable = false)

Catalyst流程解析

Catalyst在論文中被叫做優化器（Optimizer），這部分是論文裡面較為核心的內容，不過其實流程還是蠻好理解的，依舊貼下論文裡面的圖。

主要流程大概可以分為以下幾步：

1. Sql語句經過Antlr4解析，生成Unresolved Logical Plan（有使用過Antlr4的童鞋肯定對這一過程不陌生）
2. analyzer與catalog進行繫結（catlog儲存元資料）,生成Logical Plan;
3. optimizer對Logical Plan優化,生成Optimized LogicalPlan;
4. SparkPlan將Optimized LogicalPlan轉換成 Physical Plan;
5. prepareForExecution()將 Physical Plan 轉換成 executed Physical Plan;
6. execute()執行可執行物理計劃，得到RDD;

提前說一下吧，上述流程多數是在org.apache.spark.sql.execution.QueryExecution這個類裡面，這個貼一下簡單的程式碼，看看就好，先不多做深究。後面的文章會詳細介紹這裡的內容。

class QueryExecution(val sparkSession: SparkSession, val logical: LogicalPlan) {

  ......其他程式碼
  
  //analyzer階段
  lazy val analyzed: LogicalPlan = {
    SparkSession.setActiveSession(sparkSession)
    sparkSession.sessionState.analyzer.executeAndCheck(logical)
  }


  //optimizer階段
  lazy val optimizedPlan: LogicalPlan = sparkSession.sessionState.optimizer.execute(withCachedData)
  
  //SparkPlan階段
  lazy val sparkPlan: SparkPlan = {
    SparkSession.setActiveSession(sparkSession)
    // TODO: We use next(), i.e. take the first plan returned by the planner, here for now,
    //       but we will implement to choose the best plan.
    planner.plan(ReturnAnswer(optimizedPlan)).next()
  }

  //prepareForExecution階段
  // executedPlan should not be used to initialize any SparkPlan. It should be
  // only used for execution.
  lazy val executedPlan: SparkPlan = prepareForExecution(sparkPlan)

  //execute階段
  /** Internal version of the RDD. Avoids copies and has no schema */
  lazy val toRdd: RDD[InternalRow] = executedPlan.execute()

  ......其他程式碼
}

值得一提的是每個階段都使用了lazy懶載入，對這塊感興趣可以看看我之前的文章Scala函數語言程式設計（六） 懶載入與Stream。

上述主要介紹Spark SQL模組內容，其出現的背景以及主要解決問題。而後簡單介紹下Dataframe API的內容，以及Spark SQL解析SQL的內部框架Catalyst。後續主要會介紹Catalyst中各個步驟的流程，結合原始碼來做一些分析。

以