寫了有相當長一段時間的SparkSQL了，所以結合網上其他大神的分析，寫一篇文章，談談我對SparkSQL整個運行流程的一個簡單的理解，哈哈，畢竟程式員要做到知其然，還要知其所以然不是， SparkSQL的核心是Catalyst，SQL陳述句的決議以及最終執行計劃的運行都是Catalyst來實作的，所以對SparkSQL的學習就是對Catalyst的學習，

SparkSQL(Catalys)整體流程介紹：

無論是使用 SQL陳述句還是直接使用 DataFrame 或者 DataSet 算子，都會經過Catalyst一系列的分析和優化，最終轉換成高效的RDD的操作，主要流程如下：

1. sqlParser 決議 SQL，生成 Unresolved Logical Plan（未決議的邏輯計劃）
2. 由 Analyzer 結合 Catalog 資訊生成 Resolved Logical Plan（決議的邏輯計劃）
3. Optimizer根據預先定義好的規則(RBO)，對 Resolved Logical Plan 進行優化并生成 Optimized Logical Plan（優化后的邏輯計劃）
4. Query Planner 將 Optimized Logical Plan 轉換成多個 Physical Plan（物理計劃），然后由CBO 根據 Cost Model 算出每個 Physical Plan 的代價并選取代價最小的 Physical Plan 作為最終的 Physical Plan（最終執行的物理計劃）
5. Spark運行物理計劃，先是對物理計劃再進行進一步的優化，最終映射到RDD的操作上，和Spark Core一樣，以DAG圖的方式執行SQL陳述句，在最新的Spark3.0版本中，還增加了Adaptive Query Execution功能，會根據運行時資訊動態調整執行計劃從而得到更高的執行效率

整體的流程圖如下所示：

sqlParser 決議 SQL陳述句，得到Unresolved Logical Plan：

主要是借助于Antlr4這個強大的組件，識別SQL陳述句中的關鍵詞，然后把SQL陳述句決議成一顆語法樹，例如SQL陳述句： SELECT * FROM (SELECT * FROM src) a join (select * from src)b on a.key=b.key決議之后，會得到如下這樣的一棵樹：

樹中的每一個節點的資料結構是LogicalPlan，它是TreeNode類的子類，TreeNode類是非常基礎且重要的一個類，語法樹中的各類節點的具體實作類都是它的子類，TreeNode類內部定義遍歷樹的方法，LogicalPlan繼承了TreeNode，它增加了一些對于運算式Expression操作的方法，Expression是指不需要執行引擎計算，而可以直接計算或處理的節點，例如Cast操作、Porjection操作、四則運算和邏輯運算子運算等等，

上圖中的BinaryNode(二元節點)、 UnaryNode(一元節點)、 Leaf Node(葉子節點)是LogicalPlan的子類，它們分別代表了不同型別的操作，例如Join對應的是BinaryNode，Filter對應的是UnaryNode，AddJarCommand/AddFileCommand對應的是葉子節點等等...

決議成樹的代碼有點復雜...以后時間再研究了，這個不是當前我要關心的重點，接下來的Analyse、Optimizer、SparkPlan等等后續操作都在這個qe物件中進行了，qe物件全稱是QueryExecution：

Analyzer 結合 Catalog 資訊生成 Resolved Logical Plan：

Analyzer會使用事先定義好的一些規則(Rule)以及Catalog 等資訊對 Unresolved Logical Plan 進行決議，決議的目的是確定表對應的欄位是否存在，欄位型別是啥，資料存盤的具體位置等等， Rule也是用來決議SQL陳述句中的一些資訊，所有的Rule如下所示：

lazy val batches: Seq[Batch] = Seq(
  Batch("Hints", fixedPoint,
    new ResolveHints.ResolveBroadcastHints(conf),
    ResolveHints.ResolveCoalesceHints,
    ResolveHints.RemoveAllHints),
  Batch("Simple Sanity Check", Once,
    LookupFunctions),
  Batch("Substitution", fixedPoint,
    CTESubstitution,
    WindowsSubstitution,
    EliminateUnions,
    new SubstituteUnresolvedOrdinals(conf)),
  Batch("Resolution", fixedPoint,
    ResolveTableValuedFunctions ::
    ResolveRelations ::
    ResolveReferences ::
    ResolveCreateNamedStruct ::
    ResolveDeserializer ::        //決議反序列化方式
    ResolveNewInstance ::
    ResolveUpCast ::
    ResolveGroupingAnalytics ::
    ResolvePivot ::        //決議pivot函式
    ResolveOrdinalInOrderByAndGroupBy ::
    ResolveAggAliasInGroupBy ::
    ResolveMissingReferences ::
    ExtractGenerator ::
    ResolveGenerate ::
    ResolveFunctions ::
    ResolveAliases ::
    ResolveSubquery ::
    ResolveSubqueryColumnAliases ::
    ResolveWindowOrder ::
    ResolveWindowFrame ::
    ResolveNaturalAndUsingJoin ::
    ResolveOutputRelation ::
    ExtractWindowExpressions ::
    GlobalAggregates ::
    ResolveAggregateFunctions ::
    TimeWindowing ::
    ResolveInlineTables(conf) ::
    ResolveHigherOrderFunctions(catalog) ::
    ResolveLambdaVariables(conf) ::
    ResolveTimeZone(conf) ::
    ResolveRandomSeed ::
    TypeCoercion.typeCoercionRules(conf) ++
    extendedResolutionRules : _*),
  Batch("Post-Hoc Resolution", Once, postHocResolutionRules: _*),
  Batch("Nondeterministic", Once,
    PullOutNondeterministic),
  Batch("UDF", Once,
    HandleNullInputsForUDF),        //決議UDF
  Batch("FixNullability", Once,
    FixNullability),
  Batch("Subquery", Once,
    UpdateOuterReferences),      
  Batch("Cleanup", fixedPoint,
    CleanupAliases)
)

多個類似的Rule是封裝在同一個Batch中的，每個Batch會被執行一次或者多次，停止執行Batch的條件有兩個：一是在執行一定次數后發現Plan沒有變化，二是執行次數達到了一定的上限，這些Rule在真正應用時，是在RuleExecutor.scala類中執行的，在execute方法里就是遍歷這些Batchs，將所有的規則應用到LogicalPlan上：

這些Rule是如何應用到LogicalPlans上的，具體的實作在各個Rule實作類的apply方法中，太多了......

Optimizer 對 Resolved Logical Plan進行優化：

Optimizer的執行流程和上面Analyzer的執行流程幾乎一模一樣，這個階段的優化器主要是基于啟發式規則（Rule-based Optimizer，簡稱 RBO）對LogicalPlan進行優化，規則舉例有：列裁剪、謂詞下推、常量累加等等，具體的Rule在Optimizer.scala的 defaultBatches 變數中，執行時也是在RuleExecutor中執行的，規則太多了，就不列舉了...

Physical Plan生成：

經過上述一系列處理之后，接下來就是生成真正可以執行的Physical Plans，但是Spark會生成多個Physical Plans，然后再通過代價模型（Cost Model）得到"最優的"物理計劃：

但是代碼的注釋里又明確的有說：
// 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.
從原始碼注釋中可以看到，實際上選取的是這些Physical Plans中的第一個...所以并不一定是最優的那個

LogicalPlan生成Physical Plan的時候，是經過下面一些Strategy的優化之后得到的，例如下面的JoinSelection，內部就是選擇Join方式再運行時，到底是使用BHJ、SHJ、SMJ中的哪一種：

override def strategies: Seq[Strategy] =
  experimentalMethods.extraStrategies ++
    extraPlanningStrategies ++ (
    PythonEvals ::
    DataSourceV2Strategy ::
    FileSourceStrategy ::
    DataSourceStrategy(conf) ::
    SpecialLimits ::
    Aggregation ::
    Window ::
    JoinSelection ::
    InMemoryScans ::
    BasicOperators :: Nil)

最后的最后，我們就得到了一個比較高效的PhysicalPlan(資料結構的實作類是SparkPlan)，準備拿著這個物理計劃開始運行了，

Physical Plan的執行：

經過上述一系列處理之后，接下來就是正式的進行物理計劃的執行了，但是執行之前還要先執行prepareForExecution方法，進一步做一些優化作業，然后才是真正的去執行RDD，吐槽下，SparkSQL優化作業真的是太多了，真是厲害佩服！具體有如下這些優化規則：

/** A sequence of rules that will be applied in order to the physical plan before execution. */
protected def preparations: Seq[Rule[SparkPlan]] = Seq(
  PlanSubqueries(sparkSession),//生成子查詢
  EnsureRequirements(sparkSession.sessionState.conf),//根據磁區數量是否變化，適時的插入Shuffle操作
  CollapseCodegenStages(sparkSession.sessionState.conf),//全代碼生成，將多個操作放在同一個方法中執行，減少函式的呼叫
  ReuseExchange(sparkSession.sessionState.conf),//重復使用Shuffle資料
  ReuseSubquery(sparkSession.sessionState.conf))//重復使用子查詢結果

/**
 * 使用上面這些規則，對SparkPlan進行進一步的優化
 * Prepares a planned [[SparkPlan]] for execution by inserting shuffle operations and internal
 * row format conversions as needed.
 */
protected def prepareForExecution(plan: SparkPlan): SparkPlan = {
  preparations.foldLeft(plan) { case (sp, rule) => rule.apply(sp) }
}

優化之后，SparkPlan就真的真的是最最終待執行的計劃了(當然，Spark3.0中的AQE自適應再執行的程序中還會優化一波...)，SparkPlan樹中的節點都是***Exec后綴的節點，***Exec就是各種操作的具體實作了：