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路徑下。

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

case logical.Project(projectList, child) =>
  execution.Project(projectList, planLater(child)) :: Nil
case logical.Filter(condition, child) =>
  execution.Filter(condition, planLater(child)) :: Nil
case logical.Aggregate(group, agg, child) =>
  execution.Aggregate(partial = false, group, agg, planLater(child))(sqlContext) :: Nil
case logical.Sample(fraction, withReplacement, seed, child) =>
  execution.Sample(fraction, withReplacement, seed, planLater(child)) :: Nil

protected case class Batch(name: String, strategy: Strategy, rules: Rule[TreeType]*)

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官方提出了Hive onSpark的JIRA。Shark結束之後，拆分為兩個方向：

HiveQl.parseSql()把hql解析成logicalPlan。解析過程，提取出一些command，包括：

²  set key=value

²  cache table

²  uncache table

²  add jar

²  add file

²  dfs

²  source

然後由Hive的ParseDriver把hql解析成AST，得到ASTNode，

def getAst(sql: String): ASTNode = ParseUtils.findRootNonNullToken((new ParseDriver).parse(sql))

把Node轉化為Catalyst的LogicalPlan，轉化邏輯較複雜，也是Sparksql對hql支持的最關鍵部分。詳見HiveQl.nodeToPlan(node: Node):LogicalPlan方法。

大致轉換邏輯包括：

處理TOK_EXPLAIN和TOK_DESCTABLE

處理TOK_CREATETABLE，包括創建表時候一係列表的設置TOK_XXX

處理TOK_QUERY，包括TOK_SELECT，TOK_WHERE，TOK_GROUPBY，TOK_HAVING，TOK_SORTEDBY，TOK_LIMIT等等，對FROM後麵跟的語句進行nodeToRelation處理。

處理TOK_UNION

對Hive AST樹結構和表示不熟悉，所以此處略過。

在做優化前，會嚐試對之前生成的邏輯執行計劃進行createtabl操作，因為執行的hql可能是“CREATE TABLE XXX”，這部分處理在HiveMetastoreCatalog的CreateTables單例裏，繼承了Rule[LogicalPlan]。

以及PreInsertionCasts處理，也是HiveMetastoreCatalog裏的單例，繼承了Rule[LogicalPlan]。

 

之後的optimizer過程同SQLContext裏，用的是同一個Catalyst提供的Optimizer類。

HiveContext繼承自SQLContext，其QueryExecution也繼承自SQLContext的QueryExecution。後續執行計劃優化、物理執行計劃翻譯、處理及執行過程同SQL的處理邏輯是一致的。

 

翻譯物理執行計劃的時候，hive planner裏製定了些特定的策略，與SparkPlanner稍有不同。

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

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

 

對於Spark SQL來說，

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

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

implicit def createSchemaRDD[A <: Product: TypeTag](rdd: RDD[A]) =
new SchemaRDD(this,
SparkLogicalPlan(ExistingRdd.fromProductRdd(rdd)))

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內定義了幾批規則：

  val batches: Seq[Batch] = Seq(
    Batch("MultiInstanceRelations", Once,
      NewRelationInstances),
    Batch("CaseInsensitiveAttributeReferences", Once,
      (if (caseSensitive) Nil else LowercaseAttributeReferences :: Nil) : _*),
    Batch("Resolution", fixedPoint,
      ResolveReferences ::
      ResolveRelations ::
      NewRelationInstances ::
      ImplicitGenerate ::
      StarExpansion ::
      ResolveFunctions ::
      GlobalAggregates ::
      typeCoercionRules :_*),
    Batch("Check Analysis", Once,
      CheckResolution),
    Batch("AnalysisOperators", fixedPoint,
      EliminateAnalysisOperators)
  )

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

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

  val batches =
    Batch("Combine Limits", FixedPoint(100),
      CombineLimits) ::
    Batch("ConstantFolding", FixedPoint(100),
      NullPropagation,
      ConstantFolding,
      LikeSimplification,
      BooleanSimplification,
      SimplifyFilters,
      SimplifyCasts,
      SimplifyCaseConversionExpressions) ::
    Batch("Filter Pushdown", FixedPoint(100),
      CombineFilters,
      PushPredicateThroughProject,
      PushPredicateThroughJoin,
      ColumnPruning) :: Nil

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

    val strategies: Seq[Strategy] =
      CommandStrategy(self) ::
      TakeOrdered ::
      PartialAggregation ::
      LeftSemiJoin ::
      HashJoin ::
      InMemoryScans ::
      ParquetOperations ::
      BasicOperators ::
      CartesianProduct ::
      BroadcastNestedLoopJoin :: Nil

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

    val batches =
      Batch("Add exchange", Once, AddExchange(self)) ::
      Batch("Prepare Expressions", Once, new BindReferences[SparkPlan]) :: Nil

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內部存儲結構如下

 *    .--------------------------- Column type ID (4 bytes)
 *    |   .----------------------- Null count N (4 bytes)
 *    |   |   .------------------- Null positions (4 x N bytes, empty if null count is zero)
 *    |   |   |     .------------- Compression scheme ID (4 bytes)
 *    |   |   |     |   .--------- Compressed non-null elements
 *    V   V   V     V   V
 *   +---+---+-----+---+---------+
 *   |   |   | ... |   | ... ... |
 *  +---+---+-----+---+---------+
 *  \-----------/ \-----------/
 *       header         body

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內。

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

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

待整理

https://parquet.io/

 

Specific Docs and Codes:

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

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

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

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

 

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

https://spark.apache.org/docs/latest/sql-programming-guide.html

 

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

 

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

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

 

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

Jdbc support branchis under going

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.

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

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

理解有偏差的地方歡迎交流討論 :)

全文完 :)