Spark SQL深度理解篇：模塊實現、代碼結構及執行流程總覽

【編者按】在2014年7月1日的Spark Summit上，Databricks宣布終止對Shark的開發，將重點放到Spark SQL上。Spark SQL將涵蓋Shark的所有特性，用戶可以從Shark 0.9進行無縫的升級。日前張包峰的博客上分享了Spark SQL各個模塊的實現情況、代碼結構、執行流程以及對Spark SQL的理解。

Catalyst

Catalyst是與Spark解耦的一個獨立庫，是一個impl-free的執行計劃的生成和優化框架。

目前與Spark Core還是耦合的，對此user郵件組里有人對此提出疑問，見mail。

以下是Catalyst較早時候的架構圖，展示的是代碼結構和處理流程。

Catalyst定位

其他系統如果想基于Spark做一些類sql、標準sql甚至其他查詢語言的查詢，需要基于Catalyst提供的解析器、執行計劃樹結構、邏輯執行計劃的處理規則體系等類體系來實現執行計劃的解析、生成、優化、映射工作。

對應上圖中，主要是左側的TreeNodelib及中間三次轉化過程中涉及到的類結構都是Catalyst提供的。至于右側物理執行計劃映射生成過程，物理執行計劃基于成本的優化模型，具體物理算子的執行都由系統自己實現。

Catalyst現狀

在解析器方面提供的是一個簡單的scala寫的sql parser，支持語義有限，而且應該是標準sql的。

在規則方面，提供的優化規則是比較基礎的(和Pig/Hive比沒有那么豐富)，不過一些優化規則其實是要涉及到具體物理算子的，所以部分規則需要在系統方那自己制定和實現(如spark-sql里的SparkStrategy)。

Catalyst也有自己的一套數據類型。

下面介紹Catalyst里幾套重要的類結構。

TreeNode體系

TreeNode是Catalyst執行計劃表示的數據結構，是一個樹結構，具備一些scala collection的操作能力和樹遍歷能力。這棵樹一直在內存里維護，不會dump到磁盤以某種格式的文件存在，且無論在映射邏輯執行計劃階段還是優化邏輯執行計劃階段，樹的修改是以替換已有節點的方式進行的。

TreeNode，內部帶一個children: Seq[BaseType]表示孩子節點，具備foreach、map、collect等針對節點操作的方法，以及transformDown(默認，前序遍歷)、transformUp這樣的遍歷樹上節點，對匹配節點實施變化的方法。

提供UnaryNode,BinaryNode, LeafNode三種trait，即非葉子節點允許有一個或兩個子節點。

TreeNode提供的是范型。

TreeNode有兩個子類繼承體系，QueryPlan和Expression。QueryPlan下面是邏輯和物理執行計劃兩個體系，前者在Catalyst里有詳細實現，后者需要在系統自己實現。Expression是表達式體系，后面章節都會展開介紹。

Tree的transformation實現：

傳入PartialFunction[TreeType,TreeType]，如果與操作符匹配，則節點會被結果替換掉，否則節點不會變動。整個過程是對children遞歸執行的。

執行計劃表示模型

邏輯執行計劃

QueryPlan繼承自TreeNode，內部帶一個output: Seq[Attribute],具備transformExpressionDown、transformExpressionUp方法。

在Catalyst中，QueryPlan的主要子類體系是LogicalPlan，即邏輯執行計劃表示。其物理執行計劃表示由使用方實現(spark-sql項目中)。

LogicalPlan繼承自QueryPlan，內部帶一個reference:Set[Attribute]，主要方法為resolve(name:String): Option[NamedeExpression]，用于分析生成對應的NamedExpression。

LogicalPlan有許多具體子類，也分為UnaryNode, BinaryNode, LeafNode三類，具體在org.apache.spark.sql.catalyst.plans.logical路徑下。

邏輯執行計劃實現

LeafNode主要子類是Command體系：

各command的語義可以從子類名字看出，代表的是系統可以執行的non-query命令，如DDL。

UnaryNode的子類：

BinaryNode的子類：

物理執行計劃

另一方面，物理執行計劃節點在具體系統里實現，比如spark-sql工程里的SparkPlan繼承體系。

物理執行計劃實現

每個子類都要實現execute()方法，大致有以下實現子類(不全)。

提到物理執行計劃，還要提一下Catalyst提供的分區表示模型。

執行計劃映射

Catalyst還提供了一個QueryPlanner[Physical <: TreeNode[PhysicalPlan]]抽象類，需要子類制定一批strategies: Seq[Strategy]，其apply方法也是類似根據制定的具體策略來把邏輯執行計劃算子映射成物理執行計劃算子。由于物理執行計劃的節點是在具體系統里實現的，所以QueryPlanner及里面的strategies也需要在具體系統里實現。

在spark-sql項目中，SparkStrategies繼承了QueryPlanner[SparkPlan]，內部制定了LeftSemiJoin, HashJoin,PartialAggregation, BroadcastNestedLoopJoin, CartesianProduct等幾種策略，每種策略接受的都是一個LogicalPlan，生成的是Seq[SparkPlan]，每個SparkPlan理解為具體RDD的算子操作。

比如在BasicOperators這個Strategy里，以match-case匹配的方式處理了很多基本算子（可以一對一直接映射成RDD算子），如下：

Expression體系

Expression，即表達式，指不需要執行引擎計算，而可以直接計算或處理的節點，包括Cast操作，Projection操作，四則運算，邏輯操作符運算等。

具體可以參考org.apache.spark.sql.expressionspackage下的類。

Rules體系

凡是需要處理執行計劃樹(Analyze過程，Optimize過程，SparkStrategy過程)，實施規則匹配和節點處理的，都需要繼承RuleExecutor[TreeType]抽象類。

RuleExecutor內部提供了一個Seq[Batch]，里面定義的是該RuleExecutor的處理步驟。每個Batch代表著一套規則，配備一個策略，該策略說明了迭代次數(一次還是多次)。

Rule[TreeType <: TreeNode[_]]是一個抽象類，子類需要復寫apply(plan: TreeType)方法來制定處理邏輯。

RuleExecutor的apply(plan: TreeType): TreeType方法會按照batches順序和batch內的Rules順序，對傳入的plan里的節點迭代處理，處理邏輯為由具體Rule子類實現。

Hive相關

Hive支持方式

Spark SQL對hive的支持是單獨的spark-hive項目，對Hive的支持包括HQL查詢、hive metaStore信息、hive SerDes、hive UDFs/UDAFs/ UDTFs，類似Shark。

只有在HiveContext下通過hive api獲得的數據集，才可以使用hql進行查詢，其hql的解析依賴的是org.apache.hadoop.hive.ql.parse.ParseDriver類的parse方法，生成Hive AST。

實際上sql和hql，并不是一起支持的。可以理解為hql是獨立支持的，能被hql查詢的數據集必須讀取自hive api。下圖中的parquet、json等其他文件支持只發生在sql環境下(SQLContext)。

Hive on Spark

Hive官方提出了Hive onSpark的JIRA。Shark結束之后，拆分為兩個方向：

從這里看，對Hive的兼容支持將轉移到Hive on Spark上，之前Shark的經驗將在Hive社區的這個支持上體現。我理解，目前SparkSQL里的那種Hive支持方式，只是為了在Spark環境下集成操縱Hive數據，它的hql執行是調用Hive客戶端Driver，跑在hadoop MR上的，本身不是Hive on Spark的實現，只是為了使用RDD間接操作Hive數據集。

所以如果想要把現有Hive任務遷移到Spark上，應該使用Shark或者等待Hive on Spark。

Spark SQL里的Hive支持不是hive on spark的實現，而更像一個讀寫Hive數據的客戶端。且其hql支持只包含hive數據，與sql環境是互相獨立的。

以上兩節是Spark SQL Hive、Shark、Hive on Spark的區別和理解。

SQL Core

Spark SQL的核心是把已有的RDD，帶上Schema信息，然后注冊成類似sql里的”Table”，對其進行sql查詢。這里面主要分兩部分，一是生成SchemaRD，二是執行查詢。

生成SchemaRDD

如果是spark-hive項目，那么讀取metadata信息作為Schema、讀取hdfs上數據的過程交給Hive完成，然后根據這倆部分生成SchemaRDD，在HiveContext下進行hql()查詢。

對于Spark SQL來說，

數據方面，RDD可以來自任何已有的RDD，也可以來自支持的第三方格式，如json file、parquet file。

SQLContext下會把帶case class的RDD隱式轉化為SchemaRDD

ExsitingRdd單例里會反射出case class的attributes，并把RDD的數據轉化成Catalyst的GenericRow，最后返回RDD[Row]，即一個SchemaRDD。這里的具體轉化邏輯可以參考ExsitingRdd的productToRowRdd和convertToCatalyst方法。

之后可以進行SchemaRDD提供的注冊table操作、針對Schema復寫的部分RDD轉化操作、DSL操作、saveAs操作等等。

Row和GenericRow是Catalyst里的行表示模型

Row用Seq[Any]來表示values，GenericRow是Row的子類，用數組表示values。Row支持數據類型包括Int, Long, Double, Float, Boolean, Short, Byte, String。支持按序數(ordinal)讀取某一個列的值。讀取前需要做isNullAt(i: Int)的判斷。

各自都有Mutable類，提供setXXX(i: int, value: Any)修改某序數上的值。

層次結構

下圖大致對比了Pig，Spark SQL，Shark在實現層次上的區別，僅做參考。

查詢流程

SQLContext里對sql的一個解析和執行流程：

1.  第一步parseSql(sql: String)，simple sql parser做詞法語法解析，生成LogicalPlan。

2.  第二步analyzer(logicalPlan)，把做完詞法語法解析的執行計劃進行初步分析和映射，

目前SQLContext內的Analyzer由Catalyst提供，定義如下：

new Analyzer(catalog, EmptyFunctionRegistry, caseSensitive =true)

catalog為SimpleCatalog，catalog是用來注冊table和查詢relation的。

而這里的FunctionRegistry不支持lookupFunction方法，所以該analyzer不支持Function注冊，即UDF。

Analyzer內定義了幾批規則：

3.  從第二步得到的是初步的logicalPlan，接下來第三步是optimizer(plan)。

Optimizer里面也是定義了幾批規則，會按序對執行計劃進行優化操作。

4.  優化后的執行計劃，還要丟給SparkPlanner處理，里面定義了一些策略，目的是根據邏輯執行計劃樹生成最后可以執行的物理執行計劃樹，即得到SparkPlan。

5.  在最終真正執行物理執行計劃前，最后還要進行兩次規則，SQLContext里定義這個過程叫prepareForExecution，這個步驟是額外增加的，直接new RuleExecutor[SparkPlan]進行的。

6.  最后調用SparkPlan的execute()執行計算。這個execute()在每種SparkPlan的實現里定義，一般都會遞歸調用children的execute()方法，所以會觸發整棵Tree的計算。

其他特性

內存列存儲

SQLContext下cache/uncache table的時候會調用列存儲模塊。

該模塊借鑒自Shark，目的是當把表數據cache在內存的時候做行轉列操作，以便壓縮。

實現類

InMemoryColumnarTableScan類是SparkPlan LeafNode的實現，即是一個物理執行計劃。傳入一個SparkPlan(確認了的物理執行計)和一個屬性序列，內部包含一個行轉列、觸發計算并cache的過程(且是lazy的)。

ColumnBuilder針對不同的數據類型(boolean, byte, double, float, int, long, short, string)由不同的子類把數據寫到ByteBuffer里，即包裝Row的每個field，生成Columns。與其對應的ColumnAccessor是訪問column，將其轉回Row。

CompressibleColumnBuilder和CompressibleColumnAccessor是帶壓縮的行列轉換builder，其ByteBuffer內部存儲結構如下

CompressionScheme子類是不同的壓縮實現

都是scala實現的，未借助第三方庫。不同的實現，指定了支持的column data類型。在build()的時候，會比較每種壓縮，選擇壓縮率最小的（若仍大于0.8就不壓縮了）。

這里的估算邏輯，來自子類實現的gatherCompressibilityStats方法。

Cache邏輯

cache之前，需要先把本次cache的table的物理執行計劃生成出來。

在cache這個過程里，InMemoryColumnarTableScan并沒有觸發執行，但是生成了以InMemoryColumnarTableScan為物理執行計劃的SparkLogicalPlan，并存成table的plan。

其實在cache的時候，首先去catalog里尋找這個table的信息和table的執行計劃，然后會進行執行（執行到物理執行計劃生成），然后把這個table再放回catalog里維護起來，這個時候的執行計劃已經是最終要執行的物理執行計劃了。但是此時Columner模塊相關的轉換等操作都是沒有觸發的。

真正的觸發還是在execute()的時候，同其他SparkPlan的execute()方法觸發場景是一樣的。

Uncache邏輯

UncacheTable的時候，除了刪除catalog里的table信息之外，還調用了InMemoryColumnarTableScan的cacheColumnBuffers方法，得到RDD集合，并進行了unpersist()操作。cacheColumnBuffers主要做了把RDD每個partition里的ROW的每個Field存到了ColumnBuilder內。

UDF（暫不支持）

如前面對SQLContext里Analyzer的分析，其FunctionRegistry沒有實現lookupFunction。

在spark-hive項目里，HiveContext里是實現了FunctionRegistry這個trait的，其實現為HiveFunctionRegistry，實現邏輯見org.apache.spark.sql.hive.hiveUdfs

Parquet支持

待整理

http://parquet.io/

Specific Docs and Codes:

https://github.com/apache/incubator-parquet-format

https://github.com/apache/incubator-parquet-mr

http://www.slideshare.net/julienledem/parquet-hadoop-summit-2013

JSON支持

SQLContext下，增加了jsonFile的讀取方法，而且目前看，代碼里實現的是hadoop textfile的讀取，也就是這份json文件應該是在HDFS上的。具體這份json文件的載入，InputFormat是TextInputFormat，key class是LongWritable，value class是Text，最后得到的是value部分的那段String內容，即RDD[String]。

除了jsonFile，還支持jsonRDD，例子：

http://spark.apache.org/docs/latest/sql-programming-guide.html#json-datasets

讀取json文件之后，轉換成SchemaRDD。JsonRDD.inferSchema(RDD[String])里有詳細的解析json和映射出schema的過程，最后得到該json的LogicalPlan。

Json的解析使用的是FasterXML/jackson-databind庫，GitHub地址，wiki

把數據映射成Map[String, Any]

Json的支持豐富了Spark SQL數據接入場景。

JDBC支持

Jdbc support branchis under going

SQL92

Spark SQL目前的SQL語法支持情況見SqlParser類。目標是支持SQL92？？

1. 基本應用上，sql server 和oracle都遵循sql 92語法標準。

2. 實際應用中大家都會超出以上標準，使用各家數據庫廠商都提供的豐富的自定義標準函數庫和語法。

3. 微軟sql server的sql 擴展叫T-SQL(Transcate SQL).

4. Oracle 的sql 擴展叫PL-SQL.

存在問題

大家可以跟進社區郵件列表，后續待整理。

http://apache-spark-developers-list.1001551.n3.nabble.com/sparkSQL-thread-safe-td7263.html 

http://apache-spark-user-list.1001560.n3.nabble.com/Supported-SQL-syntax-in-Spark-SQL-td9538.html

總結

以上整理了對Spark SQL各個模塊的實現情況，代碼結構，執行流程以及自己對Spark SQL的理解。

原文鏈接： 整理對Spark SQL的理解 （責編/魏偉）

---

免費訂閱“CSDN云計算”微信公眾號，實時掌握第一手云中消息！

CSDN作為國內最專業的云計算服務平臺，提供云計算、大數據、虛擬化、數據中心、OpenStack、CloudStack、Hadoop、Spark、機器學習、智能算法等相關云計算觀點，云計算技術，云計算平臺，云計算實踐，云計算產業資訊等服務。

---