Flink table&Sql中使用Calcite

 

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. Sql Parser: 將sql語句通過java cc解析成AST(語法樹),在calcite中用SqlNode表示AST;
2. Sql Validator: 結合數字字典(catalog)去驗證sql語法；
3. 生成Logical Plan: 將sqlNode表示的AST轉換成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的各種算子。

而如果是通過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

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

將數據源注冊成表

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

// 生成一個唯一性表名 val name = createUniqueTableName() 
//生成表的 scheme val (fieldNames, fieldIndexes) = getFieldInfo[T](dataStream.getType) 
//傳入dataStream, 創建calcite可以識別的表 
val dataStreamTable = new DataStreamTable[T]( 
      dataStream, 
      fieldIndexes, 
      fieldNames, None, None ) 
//在數字字典里頭注冊該表 registerTableInternal(name, dataStreamTable)

上面函數實現的最后會調用scan，這里頭會創建一個CatalogNode對象，里頭攜帶了可以查找到數據源的表路徑。其實它是Flink 邏輯樹上的一個葉節點。

生成Flink 自身的邏輯計划

val tab = orderA.groupBy('user).select('user, 'amount.sum)
      .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節點。

//-----Filter的construct創建Calcite 的 LogicalFilter節點---- 
    //先遍歷子節點 
    child.construct(relBuilder) 
    //創建LogicalFilter 
    relBuilder.filter(condition.toRexNode(relBuilder)) 
    
//-----Project的construct創建Calcite的LogicalProject節點---- 
   //先遍歷子節點 
    child.construct(relBuilder) 
   //創建LogicalProject 
    relBuilder.project( 
      projectList.map(_.toRexNode(relBuilder)).asJava, 
      projectList.map(_.name).asJava, 
      true) 
      
//-----Aggregate的construct創建Calcite的LogicalAggregate節點---- 
    child.construct(relBuilder) 
    relBuilder.aggregate( 
  relBuilder.groupKey(groupingExpressions.map(_.toRexNode(relBuilder)).asJava), 
      aggregateExpressions.map { 
        case Alias(agg: Aggregation, name, _) => agg.toAggCall(name)(relBuilder) 
        case _ => throw new RuntimeException("This should never happen.") 
      }.asJava) 
  
//-----CatalogNode的construct創建Calcite的LogicalTableScan節點---- 
    relBuilder.scan(tablePath.asJava)

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

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

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

// 去除關聯子查詢 
    val decorPlan = RelDecorrelator.decorrelateQuery(relNode) 
    // 轉換time的標識符,比如存在rowtime標識的話,我們將會引入TimeMaterializationSqlFunction operator, 
    //這個operator我們會在codeGen中會用到 
    val convPlan = RelTimeIndicatorConverter.convert(decorPlan, getRelBuilder.getRexBuilder) 
    // 規范化logica計划,比如一個Filter它的過濾條件都是true的話,那么我們可以直接將這個filter去掉 
    val normRuleSet = getNormRuleSet
    val normalizedPlan = if (normRuleSet.iterator().hasNext) { 
      runHepPlanner(HepMatchOrder.BOTTOM_UP, normRuleSet, convPlan, convPlan.getTraitSet) 
    } else { 
      convPlan
    } 
    // 優化邏輯計划,調整節點間的上下游到達優化計算邏輯的效果,同時將 
    //節點轉換成派生於FlinkLogicalRel的節點 
    val logicalOptRuleSet = getLogicalOptRuleSet
    //用FlinkConventions.LOGICAL替換traitSet，表示轉換后的樹節點要求派生與接口 
    // FlinkLogicalRel 
    val logicalOutputProps = relNode.getTraitSet.replace(FlinkConventions.LOGICAL).simplify() 
    val logicalPlan = if (logicalOptRuleSet.iterator().hasNext) { 
      runVolcanoPlanner(logicalOptRuleSet, normalizedPlan, logicalOutputProps) 
    } else { 
      normalizedPlan
    } 
    // 將優化后的邏輯計划轉換成Flink的物理計划,同時將 
    //節點轉換成派生於DataStreamRel的節點 
    val physicalOptRuleSet = getPhysicalOptRuleSet
    val physicalOutputProps = relNode.getTraitSet.replace(FlinkConventions.DATASTREAM).simplify() 
    val physicalPlan = if (physicalOptRuleSet.iterator().hasNext) { 
      runVolcanoPlanner(physicalOptRuleSet, logicalPlan, physicalOutputProps) 
    } else { 
      logicalPlan
    }

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

public class TableScanRule extends RelOptRule { 
  //~ Static fields/initializers --------------------------------------------- 
  public static final TableScanRule INSTANCE = new TableScanRule(); 
  //~ Constructors ----------------------------------------------------------- 
  private TableScanRule() { 
    super(operand(LogicalTableScan.class, any())); 
  } 
  //默認返回True, 可以繼承matches,里面實現邏輯是判斷是否進行轉換調用onMatch 
  @Override 
  public boolean matches(RelOptRuleCall call) { 
    return super.matches(call); 
  } 
  //~ Methods ---------------------------------------------------------------- 
  //對當前節點進行轉換 
  public void onMatch(RelOptRuleCall call) { 
    final LogicalTableScan oldRel = call.rel(0); 
    RelNode newRel = 
        oldRel.getTable().toRel( 
            RelOptUtil.getContext(oldRel.getCluster())); 
    call.transformTo(newRel); 
  } 
}

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

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

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

生成Flink 可以執行的計划

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

== Physical Execution Plan == 
Stage 1 : Data Source
    content : collect elements with CollectionInputFormat
Stage 2 : Operator content : from: (user, product, amount) 
        ship_strategy : REBALANCE 
Stage 3 : Operator content : where: (<(user, 2)), select: (user, amount) 
            ship_strategy : FORWARD 
Stage 4 : Operator content : groupBy: (user), select: (user, SUM(amount) AS TMP_0) 
                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，感謝。