Apache Calcite是什麼東東

Apache Calcite面向Hadoop新的sql引擎，它提供了標準的SQL語言、多種查詢優化和連線各種資料來源的能力。除此之外，Calcite還提供了OLAP和流處理的查詢引擎。它2013年成為了Apache孵化專案以來，在Hadoop中越來越引人注目，並被眾多專案整合。比如Flink/Storm/Drill/Phoenix都依賴它做sql解析和優化。

Flink 結合 Calcite

Flink Table API&SQL 為流式資料和靜態資料的關係查詢保留統一的介面，而且利用了Calcite的查詢優化框架和SQL parser。該設計是基於Flink已構建好的API構建的，DataStream API 提供低延時高吞吐的流處理能力而且就有exactly-once語義而且可以基於event-time進行處理。而且DataSet擁有穩定高效的記憶體運算元和流水線式的資料交換。Flink的core API和引擎的所有改進都會自動應用到Table API和SQL上。
一條stream sql從提交到calcite解析、優化最後到flink引擎執行，一般分為以下幾個階段:

1 1. Sql Parser: 將sql語句通過java cc解析成AST(語法樹),在calcite中用SqlNode表示AST;
2 2. Sql Validator: 結合數字字典(catalog)去驗證sql語法；
3 3. 生成Logical Plan: 將sqlNode表示的AST轉換成LogicalPlan, 用relNode表示;
4 4. 生成 optimized LogicalPlan: 先基於calcite rules 去優化logical Plan,
5 再基於flink定製的一些優化rules去優化logical Plan；
6 5. 生成Flink PhysicalPlan: 這裡也是基於flink裡頭的rules將，將optimized LogicalPlan轉成成Flink的物理執行計劃；
 
7 6. 將物理執行計劃轉成Flink ExecutionPlan: 就是呼叫相應的tanslateToPlan方法轉換和利用CodeGen超程式設計成Flink的各種運算元。

而如果是通過table api來提交任務的話，也會經過calcite優化等階段，基本流程和直接執行sql類似:

1. table api parser: flink會把table api表達的計算邏輯也表示成一顆樹，用treeNode去表式;
在這棵樹上的每個節點的計算邏輯用Expression來表示。
2. Validate: 會結合數字字典(catalog)將樹的每個節點的Unresolved Expression進行繫結，生成Resolved Expression；
 
3. 生成Logical Plan: 依次遍歷數的每個節點，呼叫construct方法將原先用treeNode表達的節點轉成成用calcite 內部的資料結構relNode 來表達。即生成了LogicalPlan, 用relNode表示;
4. 生成 optimized LogicalPlan: 先基於calcite rules 去優化logical Plan,
再基於flink定製的一些優化rules去優化logical Plan；
5. 生成Flink PhysicalPlan: 這裡也是基於flink裡頭的rules將，將optimized LogicalPlan轉成成Flink的物理執行計劃；
6. 將物理執行計劃轉成Flink ExecutionPlan: 就是呼叫相應的tanslateToPlan方法轉換和利用CodeGen超程式設計成Flink的各種運算元。

所以在flink提供兩種API進行關係型查詢，Table API 和 SQL。這兩種API的查詢都會用包含註冊過的Table的catalog進行驗證，除了在開始階段從計算邏輯轉成logical plan有點差別以外，之後都差不多。同時在stream和batch的查詢看起來也是完全一樣。只不過flink會根據資料來源的性質(流式和靜態)使用不同的規則進行優化, 最終優化後的plan轉傳成常規的Flink DataSet 或 DataStream 程式。所以我們下面統一用table api來舉例講解flink是如何用calcite做解析優化，再轉換成回DataStream。

Table api任務的解析執行過程

Table Example

 1 // set up execution environment 
 2 val env = StreamExecutionEnvironment.getExecutionEnvironment
 3 val tEnv = TableEnvironment.getTableEnvironment(env) 
 4 //定義資料來源 
 5 val dataStream = env.fromCollection(Seq( Order(1L, "beer", 3), Order(1L, "diaper", 4), Order(3L, "rubber", 2))) 
 6 //將DataStream 轉換成 table，就是將資料來源在TableEnvironment中註冊成表 
 7 val orderA = dataStream.toTable(tEnv) 
 8 //用table api執行業務邏輯, 生成tab裡頭包含了flink 自己的logicalPlan,用LogicalNode表示 
 9 val tab = orderA.groupBy('user).select('user, 'amount.sum) 
10       .filter('user < 2L) 
11 //將table轉成成DataStream, 這裡頭就是涉及到我們calcite邏輯計劃生成 
12 // 優化、轉成可可執行的flink 運算元等過程 
13 val result = tab.toDataStream[Order]

將資料來源註冊成表

將DataStream 轉換成table的過程，其實就是將DataStream在TableEnvironment中註冊成表的過程中，主要是通過呼叫tableEnv.fromDataStream方法完成。

1 // 生成一個唯一性表名 val name = createUniqueTableName() 
2 //生成表的 scheme val (fieldNames, fieldIndexes) = getFieldInfo[T](dataStream.getType) 
3 //傳入dataStream, 建立calcite可以識別的表 
4 val dataStreamTable = new DataStreamTable[T]( 
5       dataStream, 
6       fieldIndexes, 
7       fieldNames, None, None ) 
8 //在數字字典裡頭註冊該表 registerTableInternal(name, dataStreamTable)

上面函式實現的最後會呼叫scan，這裡頭會建立一個CatalogNode物件，裡頭攜帶了可以查詢到資料來源的表路徑。其實它是Flink 邏輯樹上的一個葉節點。

生成Flink 自身的邏輯計劃

1 val tab = orderA.groupBy('user).select('user, 'amount.sum)
2       .filter('user < 2L) 

上面每次呼叫table api，就會生成Flink 邏輯計劃的節點。比如grouBy和select的呼叫會生成節點Project、Aggregate、Project，而filter的呼叫會生成節點Filter。這些節點的邏輯關係，就會組成下圖的一個Flink 自身資料結構表達的一顆邏輯樹：

因為這個例子很簡單，節點都沒有兩個子節點。這裡的實現可能有的人會奇怪，filter函式的形參型別是Expression，而我們傳進去的是"'user<2L"，是不是不對呀？ 其實這是scala比較牛逼的特性：隱式轉換，這些傳遞的表示式會先自動轉換成Expression。這些隱式轉換的定義基本都在介面類ImplicitExpressionOperations裡頭。其中user前面定義的'符號，則scala會將user字串轉化成Symbol型別。通過隱式轉換"'user<2L"表示式會生成一個LessThan物件，它會有兩個孩子Expression，分別是UnresolvedFieldReference("user")和Liter("2")。這個LessThan物件會作為Filter物件的condition。

Flink 自身的邏輯計劃 轉換成calcite可識別的邏輯計劃

根據上面分析我們只是生成了Flink的 logical Plan，我們必須將它轉換成calcite的logical Plan，這樣我們才能用到calcite強大的優化規則。在Flink裡頭會由上往下一次呼叫各個節點的construct方法，將Flink節點轉換成calcite的RelNode節點。

 1 //-----Filter的construct建立Calcite 的 LogicalFilter節點---- 
 2     //先遍歷子節點 
 3     child.construct(relBuilder) 
 4     //建立LogicalFilter 
 5     relBuilder.filter(condition.toRexNode(relBuilder)) 
 6     
 7 //-----Project的construct建立Calcite的LogicalProject節點---- 
 8    //先遍歷子節點 
 9     child.construct(relBuilder) 
10    //建立LogicalProject 
11     relBuilder.project( 
12       projectList.map(_.toRexNode(relBuilder)).asJava, 
13       projectList.map(_.name).asJava, 
14       true) 
15       
16 //-----Aggregate的construct建立Calcite的LogicalAggregate節點---- 
17     child.construct(relBuilder) 
18     relBuilder.aggregate( 
19   relBuilder.groupKey(groupingExpressions.map(_.toRexNode(relBuilder)).asJava), 
20       aggregateExpressions.map { 
21         case Alias(agg: Aggregation, name, _) => agg.toAggCall(name)(relBuilder) 
22         case _ => throw new RuntimeException("This should never happen.") 
23       }.asJava) 
24   
25 //-----CatalogNode的construct建立Calcite的LogicalTableScan節點---- 
26     relBuilder.scan(tablePath.asJava)

通過以上轉換後，就生成了Calcite邏輯計劃:

優化邏輯計劃並轉換成Flink的物理計劃

這部分實現Flink統一封裝在optimize方法裡頭，這個方法具體的實現如下:

 1 // 去除關聯子查詢 
 2     val decorPlan = RelDecorrelator.decorrelateQuery(relNode) 
 3     // 轉換time的識別符號,比如存在rowtime標識的話,我們將會引入TimeMaterializationSqlFunction operator, 
 4     //這個operator我們會在codeGen中會用到 
 5     val convPlan = RelTimeIndicatorConverter.convert(decorPlan, getRelBuilder.getRexBuilder) 
 6     // 規範化logica計劃,比如一個Filter它的過濾條件都是true的話,那麼我們可以直接將這個filter去掉 
 7     val normRuleSet = getNormRuleSet
 8     val normalizedPlan = if (normRuleSet.iterator().hasNext) { 
 9       runHepPlanner(HepMatchOrder.BOTTOM_UP, normRuleSet, convPlan, convPlan.getTraitSet) 
10     } else { 
11       convPlan
12     } 
13     // 優化邏輯計劃,調整節點間的上下游到達優化計算邏輯的效果,同時將 
14     //節點轉換成派生於FlinkLogicalRel的節點 
15     val logicalOptRuleSet = getLogicalOptRuleSet
16     //用FlinkConventions.LOGICAL替換traitSet，表示轉換後的樹節點要求派生與介面 
17     // FlinkLogicalRel 
18     val logicalOutputProps = relNode.getTraitSet.replace(FlinkConventions.LOGICAL).simplify() 
19     val logicalPlan = if (logicalOptRuleSet.iterator().hasNext) { 
20       runVolcanoPlanner(logicalOptRuleSet, normalizedPlan, logicalOutputProps) 
21     } else { 
22       normalizedPlan
23     } 
24     // 將優化後的邏輯計劃轉換成Flink的物理計劃,同時將 
25     //節點轉換成派生於DataStreamRel的節點 
26     val physicalOptRuleSet = getPhysicalOptRuleSet
27     val physicalOutputProps = relNode.getTraitSet.replace(FlinkConventions.DATASTREAM).simplify() 
28     val physicalPlan = if (physicalOptRuleSet.iterator().hasNext) { 
29       runVolcanoPlanner(physicalOptRuleSet, logicalPlan, physicalOutputProps) 
30     } else { 
31       logicalPlan
32     }

這段涉及到多個階段，每個階段無非都是用Rule對邏輯計劃進行優化和改進。每個Rule的邏輯大家自己去看，如果我想自己自定義一個Rule該如何做呢？首先宣告定義於派生RelOptRule的一個類，然後再建構函式中要求傳入RelOptRuleOperand物件，該物件需要傳入你這個Rule將要匹配的節點型別。如果你的自定義的Rule只用於LogicalTableScan節點，那麼你這個operand物件應該是operand(LogicalTableScan.class, any())。就像這樣一樣

 1 public class TableScanRule extends RelOptRule { 
 2   //~ Static fields/initializers --------------------------------------------- 
 3   public static final TableScanRule INSTANCE = new TableScanRule(); 
 4   //~ Constructors ----------------------------------------------------------- 
 5   private TableScanRule() { 
 6     super(operand(LogicalTableScan.class, any())); 
 7   } 
 8   //預設返回True, 可以繼承matches,裡面實現邏輯是判斷是否進行轉換呼叫onMatch 
 9   @Override 
10   public boolean matches(RelOptRuleCall call) { 
11     return super.matches(call); 
12   } 
13   //~ Methods ---------------------------------------------------------------- 
14   //對當前節點進行轉換 
15   public void onMatch(RelOptRuleCall call) { 
16     final LogicalTableScan oldRel = call.rel(0); 
17     RelNode newRel = 
18         oldRel.getTable().toRel( 
19             RelOptUtil.getContext(oldRel.getCluster())); 
20     call.transformTo(newRel); 
21   } 
22 }

通過以上程式碼對邏輯計劃進行了優化和轉換，最後會將邏輯計劃的每個節點轉換成Flink Node，既可物理計劃。整個轉換過程最後的結果如下：

1 == Optimized pyhical Plan == DataStreamGroupAggregate(groupBy=[user], select=[user, SUM(amount) AS TMP_0])  
2         
3 DataStreamCalc(select=[user, amount], where=[<(user, 2)]) 
4         
5 DataStreamScan(table=[[_DataStreamTable_0]]) 

我們發現Filter節點在樹結構中下移了，這樣對資料進行操作時現在過濾再做聚合，可以減少計算量。

生成Flink 可以執行的計劃

這一塊只要是遞迴呼叫各個節點DataStreamRel的translateToPlan方法，這個方法轉換和利用CodeGen超程式設計成Flink的各種運算元。現在就相當於我們直接利用Flink的DataSet或DataStream API開發的程式。整個流程的轉換大體就像這樣：

1 == Physical Execution Plan == 
2 Stage 1 : Data Source
3     content : collect elements with CollectionInputFormat
4 Stage 2 : Operator content : from: (user, product, amount) 
5         ship_strategy : REBALANCE 
6 Stage 3 : Operator content : where: (<(user, 2)), select: (user, amount) 
7             ship_strategy : FORWARD 
8 Stage 4 : Operator content : groupBy: (user), select: (user, SUM(amount) AS TMP_0) 
9                 ship_strategy : HASH

總結

不過這個樣例中忽略了流處理中最有趣的部分：window aggregate 和 join。這些操作如何用SQL表達呢？Apache Calcite社群提出了一個proposal來討論SQL on streams的語法和語義。社群將Calcite的stream SQL描述為標準SQL的擴充套件而不是另外的 SQL-like語言。這有很多好處，首先，熟悉SQL標準的人能夠在不學習新語法的情況下分析流資料。靜態表和流表的查詢幾乎相同，可以輕鬆地移植。此外，可以同時在靜態表和流表上進行查詢，這和flink的願景是一樣的，將批處理看做特殊的流處理(批看作是有限的流)。最後，使用標準SQL進行流處理意味著有很多成熟的工具支援

此文轉載自http://blog.chinaunix.net/uid-29038263-id-5765791.html，感謝。