Flink Table&Sql 如何结合Apache 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 5 : 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进行流处理意味着有很多成熟的工具支持。