Spark SQL原理解析前言：
Spark SQL源碼剖析（一）SQL解析框架Catalyst流程概述
Spark SQL源碼解析（二）Antlr4解析Sql並生成樹
前面已經介紹了SQL parse，將一條SQL語句使用antlr4解析成語法樹並使用訪問者模式生成Unresolved LogicalPlan，然后是Analysis階段將Unresolved LogicalPlan轉換成Resolved LogicalPlan。這一篇我們介紹Optimization階段，和生成Physical Planning階段。
經過這兩個階段后，就差不多要到最后轉換成Spark的RDD任務了。
Spark SQL Optimization階段概述
先來看看Logical Optimization階段。
上一篇我們討論了Analysis階段如何生成一個真正的Logical Plan樹。這一階段聽名字就知道是優化階段，Spark SQL中有兩個部分的優化，第一部分就是這里，是rule-base階段的優化，就是根據各種關系代數的優化規則，對生成的Logical Plan適配，匹配到就進行相應的優化邏輯。這些規則大概有：投影消除，constant folding，替換null值，布爾表達式簡化等等。當然大部分規則細節我也不是很清楚，僅僅能從名字推斷一二。這
同時還可以添加自己的優化rule，也比較容易實現，論文中就給出了一段自定義優化rule的代碼：
object DecimalAggregates extends Rule[LogicalPlan] {
/** Maximum number of decimal digits in a Long */
val MAX_LONG_DIGITS = 18
def apply(plan: LogicalPlan): LogicalPlan = {
plan transformAllExpressions {
case Sum(e @ DecimalType.Expression(prec , scale))
if prec + 10 <= MAX_LONG_DIGITS =>
MakeDecimal(Sum(UnscaledValue(e)), prec + 10, scale)
}
}
這段代碼的大意是自定義了一個rule，如果匹配到SUM的表達式，那就執行相應的邏輯，論文里描述這里是找到對應的小數並將其轉換為未縮放的64位LONG。具體邏輯看不是很明白不過不重要，重要的是編寫自己的優化rule很方便就是。
順便點一下另一種優化，名字叫做cost-base優化（CBO），是發生在Physical Planning階段的，這里就先賣個關子，后面說到的時候再討論吧。
然后看到源碼的時候，會發現Optimizer這個類也是繼承自RuleExecutor，繼承這個類之后的流程基本都是一樣的。前面分析Analysis階段的時候已經有詳細介紹過這個流程，這里就不展開說了。
其實這優化器的重點應該是各種優化規則，這里我覺得更多的是設計到關系代數表達式優化理論方面的知識，這部分我也不甚精通，所以也就不說了。對這塊感興趣的童鞋可以看看網上別人的文章，這里順便列幾個可能有幫助的博客，
下面還是來看看最開始的例子進行Optimization階段后會變成什么樣吧，先看看之前的示例代碼：
val df = Seq((1, 1)).toDF("key", "value")
df.createOrReplaceTempView("src")
val queryCaseWhen = sql("select key from src ")
然后在Optimization優化階段后，變成了：
Project [_1#2 AS key#5]
+- LocalRelation [_1#2, _2#3]
好吧，看起來沒什么變化，與Analysis階段相比，也就少了個SubqueryAlias ，符合預期。不過也對，就一條SELECT語句能優化到哪去啊。
Physical Planning生成階段概述
相比較於Logical Plan，Physical plan算是Spark可以去執行的東西了，當然本質上它也是一棵樹。
前面說到，Spark有一種cost-based的優化。主要就在這一階段，在這一階段，會生成一個或多個Physical Plan，然后使用cost model預估各個Physical Plan的處理性能，最后選擇一個最優的Physical Plan。這里最主要優化的是join操作，當觸發join操作的時候，會根據左右兩邊的數據集判斷，然后決定使用Broadcast join，還是傳統的Hash join，抑或是MergeSort join，有關這幾種join的區別這里就不詳細解釋了，有興趣童鞋可以百度看看。
除了cost-based優化，這一階段也依舊會有rule-based優化，所以說RuleExecutor這個類是很重要的，前面提到的Analysis階段也好，Optimization階段也好，包括這里的Physical Plan階段，只要是涉及到rule-based優化，都會跟RuleExecutor這個類扯上關系。當然這樣無疑是極大使用了面向對象的特性，不同的階段編寫不同的rule就行，一次編寫，到處復用。
Physical Planning源碼分析
首先是在QueryExecution中調度，
class QueryExecution(val sparkSession: SparkSession, val logical: LogicalPlan) {
......其他代碼
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()
}
......其他代碼
}
這里的planner是org.apache.spark.sql.execution.SparkPlanner這個類，而這個類繼承自org.apache.spark.sql.catalyst.planning.QueryPlanner，plan()方法也是在父類QueryPlanner中實現的。和RuleExecution類似，QueryPlanner中有一個返回Seq[GenericStrategy[PhysicalPlan]]的方法：def strategies: Seq[GenericStrategy[PhysicalPlan]]，這個方法會在子類（也就是SparkPlanner）重寫，然后被QueryPlanner的plan()方法調用。
我們來看看SparkPlanner中strategies方法的重寫，再來看QueryPlanner的plan()方法吧。
class SparkPlanner(
val sparkContext: SparkContext,
val conf: SQLConf,
val experimentalMethods: ExperimentalMethods)
extends SparkStrategies {
......其他代碼
override def strategies: Seq[Strategy] =
experimentalMethods.extraStrategies ++
extraPlanningStrategies ++ (
PythonEvals ::
DataSourceV2Strategy ::
FileSourceStrategy ::
DataSourceStrategy(conf) ::
SpecialLimits ::
Aggregation ::
Window ::
JoinSelection ::
InMemoryScans ::
BasicOperators :: Nil)
......其他代碼
strategies()返回策略列表，是生成策略GenericStrategy，這是個具體的抽象類，位於org.apache.spark.sql.catalyst.planning包。所謂生成策略，就是決定如果根據Logical Plan生成Physical Plan的策略。比如上面介紹的join操作可以生成Broadcast join，Hash join，抑或是MergeSort join，就是一種生成策略，具體的類就是上面代碼中的JoinSelection。每個生成策略GenericStrategy都是object，其apply()方法返回的是Seq[SparkPlan]，這里的SparkPlan就是PhysicalPlan（注意：下文會將SparkPlan和PhysicalPlan混着用）。
明白了生成策略后，就可以來看看QueryPlanner的plan()方法了。
abstract class QueryPlanner[PhysicalPlan <: TreeNode[PhysicalPlan]] {
......其他代碼
def plan(plan: LogicalPlan): Iterator[PhysicalPlan] = {
// Obviously a lot to do here still...
// Collect physical plan candidates.
val candidates = strategies.iterator.flatMap(_(plan)) //迭代調用並平鋪，變成Iterator[SparkPlan]
// The candidates may contain placeholders marked as [[planLater]],
// so try to replace them by their child plans.
val plans = candidates.flatMap { candidate =>
val placeholders = collectPlaceholders(candidate)
if (placeholders.isEmpty) {
// Take the candidate as is because it does not contain placeholders.
Iterator(candidate)
} else {
// Plan the logical plan marked as [[planLater]] and replace the placeholders.
placeholders.iterator.foldLeft(Iterator(candidate)) {
case (candidatesWithPlaceholders, (placeholder, logicalPlan)) =>
// Plan the logical plan for the placeholder.
val childPlans = this.plan(logicalPlan)
candidatesWithPlaceholders.flatMap { candidateWithPlaceholders =>
childPlans.map { childPlan =>
// Replace the placeholder by the child plan
candidateWithPlaceholders.transformUp {
case p if p.eq(placeholder) => childPlan
}
}
}
}
}
}
val pruned = prunePlans(plans)
assert(pruned.hasNext, s"No plan for $plan")
pruned
}
......其他代碼
}
這里的流程其實不難，主要工作其實就是調用各個生成策略GenericStrategy的apply()方法，生成Iterator[SparkPlan]。后面很大部分代碼是處理占位符，按我的理解，在生成Logical Plan的時候，可能有些無意義的占位符，這種需要使用子節點替換調它。倒數第三行prunePlans()方法按注釋說是用來去掉bad plan的，但看實際代碼只是原封不動返回。
這樣最終就得到一個Iterator[SparkPlan]，每個SparkPlan就是可執行的物理操作了。
大致流程就是如此，當然具體到一些生成策略沒有細說，包括輸入源策略，聚合策略等等，每一個都蠻復雜的，這里就不細說，有興趣可以自行查閱。
對了，最后還要看看示例代碼到這一步變成什么樣了，先上示例代碼：
//生成DataFrame
val df = Seq((1, 1)).toDF("key", "value")
df.createOrReplaceTempView("src")
//調用spark.sql
val queryCaseWhen = sql("select key from src ")
經過Physical Planning階段后，變成如下：
Project [_1#2 AS key#5]
+- LocalTableScan [_1#2, _2#3]
對比上面的optimized階段，直觀看就是LocalRelation變成LocalTableScan。變得更加具體了，但實際上，Project也變了，雖然打印名字相同，但一個的類型是Project，本質上是LogicalPlan。而一個是ProjectExec，本質上是SparkPlan（也就是PhysicalPlan）。這一點通過斷點看的更清楚。
到這一步已經很解決終點了，后面再經過一個Preparations階段就能生成RDD了，剩下的部分留待下篇介紹吧。
以上~