前言

大家好，我是土哥，

這已經是我為讀者寫的第21篇 Flink系列文章了，

上周有粉絲在群里問，在流計算平臺撰寫完Flink sql后，為什么通過一鍵提交按鈕，就可以將sql提交到yarn集群上面了？

由于現在各大廠對業務分層特別清晰，平臺方向和底層技術開發會被單獨劃分，所以好多大資料同學撰寫完 Flink Sql 后，只需通過提交按鈕將其提交到集群上，對背后的提交原理些許不太清楚，

下面土哥將為大家揭開這層神秘的面紗，挖掘 Flink Sql 背后的提交原理和原始碼設計，（硬核文章，建議收藏！）

熟悉平臺

故事

小笨豬阿土剛入職某大資料公司擔任實習生，然后主管交給阿土一個任務，讓其熟悉公司的 Flink 流計算平臺，

阿土登錄流計算平臺后，看到平臺上面可以撰寫 Sql 語法，于是就寫了一個簡單的 sql

他發現旁邊有個效驗功能、于是就點擊了一下，這時平臺彈出 SQL 語法效驗正確，阿土心中暗暗自喜，看來我的 sql 功底還是不錯嘛，

SQL 語法效驗完成后，阿土點擊提交按鈕，流計算平臺提示，SQl 語法效驗正確，已成功提交集群，

Flink sql 代碼居然提交到yarn集群上了？？？

小笨豬阿土感到很驚訝，sql 就這樣直接提交到集群了哇，這時候小笨豬的導師猴哥過來了，看到小笨豬的操作后，表揚了幾句，

阿土，完成的不錯啊，已經可以提交 sql 代碼啦，但是你可別小看這簡單的提交，這背后的門路可不淺呦，

這樣吧，你好好探索一下這個sql提交的原理，然后寫一篇分析報告，在咱們組分享一下，

啊......啊......

小笨豬阿土聽到猴哥的要求后，一下就蔫了，從此之后，阿土就和 Flink sql 走在了一起，

剛開始阿土很懵，于是就開始搜查 Flink sql 相關文章，過了幾天，終于理清了一些思路， 小笨豬將其流程總結為以下幾個點：

1. Flink Sql決議器

2. Flink Planner 和 Blink Planner

3. Blink Sql提交流程

1.1、了解Calcite

為方便用戶使用 Flink 流計算組件，Flink 社區設計了四種抽象，在這些抽象中，Sql API 屬于Flink的最上層抽象，是 Flink 的一等公民，這就方便用戶或者開發者直接通過 Sql 撰寫來提交任務，

但經過阿土的調查后 發現，Flink sql 在提交任務時，并不是向 DataStream API 那樣，直接被轉為 StreamGraph，經過優化生成 JobGraph 提交到集群的,而是需要對撰寫的 Sql 進行決議、驗證、優化等操作，在這中間，社區引入了一個強大的決議器，那就是Calcite，

阿土好好調研了一番Calcite

Calcite屬于Apache旗下的一個動態資料管理框架，具備很多資料庫管理系統的功能，它可以對SQL進行 SQL 決議，SQL 校驗，SQL 查詢優化，SQL 生成以及資料連接查詢等操作，它不存盤元資料和基本資料，不包含處理資料的演算法，而是作為一個中介的角色，將上層SQL和底層處理引擎打通，將其SQL轉為底層處理引擎需要的資料格式，

它不受上層編程語言的限制，前端可以使用 SQL、Pig、Cascading 等語言，只要通過 Calcite 提供的 SQL Api 將它們轉化成關系代數的抽象語法樹即可，并根據一定的規則和成本對抽象語法樹進行優化，最后推給各個資料處理引擎來執行，

所以 Calcite 不涉及物理規劃層，它通過擴展配接器來連接多種后端的資料源和資料處理引擎，如 Hive，Drill，Flink，Phoenix等，

1.2、Calcite執行步驟

小笨豬阿土簡單畫了一下Calcite的執行流程，主要涉及5個部分 SQL決議、SQL校驗、SQL查詢優化、SQL生成、執行等，

在這個流程中，Calcite各階段扮演的角色如下：

1. SQL決議，通過 JavaCC 實作，使用 JavaCC 撰寫 SQL 語法描述檔案，將 SQL 決議成未經校驗的 AST 語法樹，

2. SQL校驗，通過與元資料結合驗證 SQL 中的 Schema、Field、 Function 是否存在，輸入輸出型別是否匹配等，

3. SQL優化，對上個步驟的輸出( RelNode ，邏輯計劃樹)進行優化，使用兩種規則：基于規則優化 和 基于代價優化，得到優化后的物理執行計劃，

4. SQL生成，將物理執行計劃生成為在特定平臺/引擎的可執行程式，如生成符合 MySQL 或 Oracle 等不同平臺規則的 SQL 查詢陳述句等，

5. 執行，執行是通過各個執行平臺執行查詢，得到輸出結果，

其中，Calcite再與其他處理引擎結合時，到SQL優化階段就已經結束，所以流程圖簡化為：

阿土看完Calcite的原理后，開始想，那Calcite是怎么在Flink中扮演的角色呢？

這時猴哥過來給阿土說，單純的看一些理論文章，是搞不清楚底層設計實作的，阿土啊，你可以看看原始碼，

聽了猴哥的一番話后，阿土開始啃起了Flink1.13.2的Flink Sql原始碼

2.1 Flink Planner和Blink Planner

在1.9.0版本以前，社區使用Flink Planner作為查詢處理器，通過與Calcite進行連接，為Table/SQL API提供完整的決議、優化和執行環境，使其SQL被轉為DataStream API的 Transformation，然后再經過StreamJraph -> JobGraph -> ExecutionGraph等一系列流程，最終被提交到集群，

在1.9.0版本，社區引入阿里巴巴的Blink，對FIink TabIe & SQL模塊做了重大的重構，保留了 Flink Planner 的同時，引入了 Blink PIanner，沒引入以前，Flink 沒考慮流批作業統一，針對流批作業，底層實作兩套代碼，引入后，基于流批一體理念，重新設計算子，以流為核心，流作業和批作業最終都會被轉為transformation，

2.2 Blink Planner與Calcite關系

在之后的版本，為了實作Flink流批一體的愿景，通過Blink Planner與Calcite進行對接,對接流程如下：

1. 在Table/SQL 撰寫完成后，通過Calcite 中的parse、validate、rel階段，以及Blink額外添加的convert階段,將其先轉為Operation；

2. 通過Blink Planner 的translateToRel、optimize、translateToExecNodeGraph和translateToPlan四個階段，將Operation轉換成DataStream API的 Transformation；

3. 再經過StreamJraph -> JobGraph -> ExecutionGraph等一系列流程，SQL最終被提交到集群，

小笨豬根據查詢后的資料以及查看Flink 1.13.2版本原始碼后，畫出如下SQL執行流程圖，

阿土根據對原始碼的分析后，發現無論是Flink SQL執行DDL操作、還是DQL操作或者DML操作、最終都可以將其總結為兩個階段：

1. SQL 陳述句到 Operation 程序，即Parse階段；

2. Operation 到 Transformations 程序，即Translate階段，

3.1、Parse階段

在Parse階段一共包含parse、validate、rel、convert部分

Calcite的 parse 決議模塊是基于javacc實作的，javacc是一個詞法分析生成器和語法分析生成器，詞法分析器于將輸入字符流決議成一個一個的token，以下面這段SQL陳述句為例:

示例1 ：

在 parse 部分，上面的SQL陳述句最后會被決議為如下一組token：

接下來語法分析器會以詞法分析器決議出來的token序列作為輸入來進行語法分析，分析程序使用遞回下降語法決議，LL(k)，

其中，第一個L表示從左到右掃描輸入；第二個L表示每次都進行最左推導(在推導語法樹的程序中每次都替換句型中最左的非終結符為終結符，類似還有最右推導)；

k表示的是每次向前探索(lookahead)k 個終結符，

分析所依賴的的詞法法則定義在一個parser.jj檔案中，

在經過詞法分析和語法分析后，一段 SQL 陳述句會被決議成一顆抽象語法樹（Abstract Syntax Tree，AST），樹的節點型別在 Calcite 中以 SqlNode 來表示，不同節點以不同子型別的SqlNode來表示，

同樣以上面的SQL為例，在這段SQL中:

1. id, score, T 等為 SqlIdentifier，表示一個欄位名或表名的識別符號;

2. select和cast()為SqlCall，表示一個行為或動作，其中cast()為一個SqlBasicCall，表示一個函式呼叫，具體呼叫的是什么函式，由其內部的SqlOperator決定，比如這里是一個二元運算子“<”，對應SqlBinaryOperator，operator的名字是“<”，類別是SqlKind.LESS_THAN;

3. int 為 SqlDataTypeSpec，表示一個型別定義;

4. 'hello'和 10 為SqlLiteral，表示一個常量;

在Calcite中，所有的操作都是一個SqlCall, 如查詢是一個 SqlSelect, 洗掉是一個 SqlDelete 等，它們都是 SqlCall 的子型別，select的查詢條件等為 SqlCall 中的引數，示例1 的 SQL 陳述句最終生成的語法樹形式如下：

如果把示例1中的直接從一個表查詢資料，改為從兩張表的關聯結果中查詢資料，例如：

示例2：

則相應的AST形式如下：

其中只有FROM子樹部分由原來的SqlIdentifier節點變成了一棵SqlJoin子樹，其他部分與示例1相同所以在圖中省略了，

校驗（validate）階段

對經過parser決議出的AST進行有效性驗證，驗證的方面主要包括以下兩方面：

1. 表名、欄位名、函式名是否正確，如在某個查詢的欄位在當前SQL位置上是否存在或有歧義（當前可見的多個資料源中同時存在該名稱的欄位）

2. 特定型別操作自身的合法性，如group by聚合中的聚合函式是否存在嵌套呼叫，使用AS重命名時，新名字是否是x.y的形式等

針對上面的第一種情況，在校驗程序中首先需要明確兩個最重要的概念：NameSpace和Scope，

NameSpace代表一個邏輯上的資料源，可以是一張表，也可以是一個子查詢，而Scope則代表了在 SQL 的某個位置，表和欄位的可見范圍，

從概念中可以看出，在某個 SQL位置上，某個欄位所對應的 scope 可能包含多個 namespace，在 validate 階段決議出來的 scope 和 namespace 資訊會被保存下來，在后面轉換成邏輯執行計劃的時候還會用到，

通過一個示例來看什么是 NameSpace 和 scope

示例3

在上面這樣一段SQL陳述句中包含四個namespace：

對于SQL中的不同運算式，根據它們所在的位置，它們所對應的scope如下：

那么在校驗第一種情況的時候，整個校驗程序的核心就在于為不同的SqlNode節點生成其對應的namespace和scope，然后對該SqlNode涉及的欄位和namespace與scope的對應關系進行校驗，

對于第二種情況的校驗，則需要根據具體的節點型別分別實作了，

在Calcite中，validator的具體實作類是SqlValidatorImpl，namespace和scope分別由介面SqlValidatorNamespace和SqlValidatorScope表示，圖中涉及到的xxxNamespace和xxxScope分別是這兩個類的子類，

下圖是從呼叫validator.validate(sqlNode)開始，對一段查詢陳述句的表名和欄位名進行校驗的時序圖，

大體程序都已經在圖中的注解里進行了說明，需要補充的一點是，在通過 emptyScope.resolve決議表名時，表資訊是通過具體的catalogReader從catalog的schema中查找出來的，

具體使用什么catalog和catalogReader，是在validator創建之初決定的，

在flink中，根據用戶的配置，catalog可能是 GenericInMemoryCatalog（基于記憶體的catalog）或HiveCatalog（基于hive metastore的catalog），

如下圖所示：

rel階段是將SqlNode組成的一棵抽象語法樹轉化為一棵由RelNode和RexNode組成的關系代數樹，或者稱為執行計劃，RelNode表示關系運算式，如投影（Project），即SELECT，和連接（JOIN）等；

RexNode表示行運算式，如示例中的 CAST(score AS INT)、T1.id < 10，

以示例2的語法樹為例，在經過rel階段轉換后會生成下圖所示的執行計劃：

rel階段只處理DML和DQL

因為DDL實際上可以認為是對元資料的修改，不涉及復雜關系查詢，也就不用進行關系代數轉換來優化執行， 所以也無需轉換為表示，根據對應的SqlNode中保存的資訊已經可以直接執行了，

在calcite中，SqlToRelConverter用于對關系運算式進行轉換，Flink中通過如下方式使用calcite將AST轉換成邏輯執行計劃，如下圖原始碼所示，

從Flink1.13.2原始碼中可以看到轉換的入口是convertQuery方法，

SqlToRelConverter中的簡單的轉換流程如下圖所示:

針對每種可能的根節點型別都有對應的轉換方法，其中DELETE、UPDATE、MERGE、WITH和VALUES這幾種語法在flink流式SQL中還不支持，并且其轉換程序也比較簡單，后文不再詳細分析，

對于一棵轉換后得到的邏輯執行計劃樹中的節點，其實在AST中都是可以一一對應的找到對應的節點的，所以轉換程序本身并不涉及很復雜的演算法，大部分程序是提取已有SqlNode節點中記錄的資訊，然后生成對應的RelNode和RexNode，并設定RelNode間的父子關系，

從圖中也可以看出在calcite里最終都會生成一個LogicalModify節點，通過節點內的operation屬性來標識不同的含義，但是目前flink支持的DML只有insert陳述句，而且并不會生成LogicalModify節點，而是直接轉換成了ModifyOperation，并在需要的時候轉換成flink內部自己定義的節點型別LogicalSink，也因為這個原因，對于DML的轉換流程圖中是略有簡化的，insert、delete、update和merge本身都可以帶查詢陳述句，因此實際轉換的時候都會遞回地先對查詢部分進行轉換，

上圖所示流程中只展示了對關系運算式的轉換，但是每個關系節點（RelNode）中的行運算式同樣需要經過轉換得來，

Calcite中行運算式的轉換依賴于兩個物件：BlackBoard和SqlNodeToRexConverter，

BlackBoard是對select進行轉換時的一個臨時作業空間，它就像一塊“黑板”一樣，可以臨時記錄下轉換程序中需要的資訊，比如select依賴的scope、當前的root節點、當前節點是否是top節點等，

BlackBoard本身還是一個shuttle，針對不同型別的SqlNode，其內部都有對應的visit方法，其中除SqlCall、SqlLiteral、SqlIntervalQualifier外，都可由BlackBoard和SqlToRelConverter中定義的各種convertXXX方法進行轉換，這三種型別的SqlNode則需要借助SqlNodeToRexConverter來進行轉換，

SqlLiteral、SqlIntervalQualifier的轉換比較簡單，就是從原來的SqlNode中提取資訊進行簡單的處理和轉換，然后生成對應的RexNode，

重點：這一步負責將RelNode tree轉換成operation

這里只有以及的查詢子句會首先通過轉換為

RelNode轉換成Operation的程序很簡單，針對四種型別的操作，其各自的轉換程序如下：·

1. CreateTable @convertCreateTable

   如果AST的根節點是SqlCreateTable，提取節點中記錄的schema、properties、comment、primary keys、if not exists資訊，創建CatalogTable物件，然后創建CreateTableOperation

2. DropTable @convertDropTable

   如果AST的根節點是SqlDropTable，提取節點中記錄的full table name、if exists資訊，創建DropTableOperation物件

3. Insert @convertInsert

   如果AST的根節點是RichSqlInsert，提取節點中記錄的目標表的完整路徑和查詢運算式，先將查詢運算式通過convertSqlQuery轉換成QueryOperation，然后以轉換后的QueryOperation為子節點創建ModifyOperation物件，

這里分兩種情況:

(1)使用SQL API執行了 insert into 陳述句，將資料寫入已經通過 TableEnvironment注冊過的表中，此時創建的是CatalogSinkModifyOperation

(2)使用Table API的toXXXStream將table物件轉換成了DataStream，創建的是OutputConversionModifyOperation

1. Query @convertSqlQuery

   如果根節點的SqlKind是SqlKind.Query，先通過FlinkPlannerImpl.rel將SqlNode轉換成RelNode，然后創建PlannerQueryOperation物件

3.2、Translate階段

在Translate階段，通過Blink Planner 的translateToRel、optimize、translateToExecNodeGraph和translateToPlan四個階段:將Operation轉換成 Transformations，

重點：

1. 從operation開始，先將ModifyOperation通過translateToRel方法轉換成Calcite RelNode邏輯計劃樹，在對應轉換成FlinkLogicalRel（RelNode邏輯計劃樹）；

2. 然后經過 呼叫optimize方法將FlinkLogicalRel 優化成FlinkPhysicalRel，

3. 再呼叫translateToExecNodeGraph方法將FlinkPhysicalRel轉為execGraph

4. 最后呼叫translateToPlan方法將execGraph轉為transformations

從邏輯計劃變成物理計劃（RelNode），

Flink1.13.2原始碼如下：

這個程序可以看成是convert: RelNode => Operation的逆程序，

邏輯也很簡單，無論是使用SQL API還是Table API，最終生成的operation的根節點一定是ModifyOperation，因為只有insert陳述句或者將Table轉換成DataStream后，在DataStream結果上面寫入sink才能觸發執行，

前文提到過ModifyOperation最終都會被轉換成flink內自定義的LogicalSink節點，該節點主要記錄資料輸出資訊，核心在于需要創建出表示資料輸出的sink，所以針對三種ModifyOperation型別分別創建sink的程序如下：

1. UnregisteredSinkModifyOperation：這個operation中直接記錄了sink資訊，因此直接提取出來創建LogicalSink即可，

2. CatalogSinkModifyOperation：根據operation中記錄的table path找到對應的table，然后根據table創建出table sink，最后使用table sink創建出LogicalSink節點，這個程序中涉及到了在catalog中決議table和使用ServiceLoader根據table資訊在classpath中查找并用于創建table sink的TableSinkFactory的程序，具體如下圖所示，

會使用兩個優化器:RBO(基于規則的優化器) 和 CBO(基于代價的優化器)

1. RBO(基于規則的優化器)會將原有運算式裁剪掉，遍歷一系列規則（Rule），只要滿足條件就轉換，生成最終的執行計劃，一些常見的規則包括磁區裁剪（Partition Prune）、列裁剪、謂詞下推（Predicate Pushdown）、投影下推（Projection Pushdown）、聚合下推、limit下推、sort下推、常量折疊（Constant Folding）、子查詢行內轉join等，

2.CBO(基于代價的優化器)會將原有運算式保留，基于統計資訊和代價模型，嘗試探索生成等價關系運算式，最終取代價最小的執行計劃，CBO的實作有兩種模型，Volcano模型，Cascades模型，這兩種模型思想很是相似，不同點在于Cascades模型一邊遍歷SQL邏輯樹，一邊優化，從而進一步裁剪掉一些執行計劃，

原始碼如下：

呼叫translateToExecNodeGraph方法將FlinkPhysicalRel轉為execGraph

呼叫translateToPlan方法將execGraph轉為transformations

通過上述四個步驟，實作將Operation轉換成 Transformations，

小笨豬通過完整的流程分析后，終于搞懂了Flink sql的決議和轉換程序，最終SQL被轉為Transformations,后面的步驟就變成了Flink DataStream的提交流程，小笨豬還是比較了解的，

分享

時間一眨眼就到了分享的那天，小笨豬在講臺上對Flink sql應用提交做了深入講解，得到了導師猴哥和主管的一致認可，