1.18.2.9.查询优化、Blink planner、解释表

电玩女神 2022-08-31 00:49 9阅读 0赞

1.18.2.9.查询优化  
1.18.2.9.1.Blink planner  
1.18.2.10.解释表

#### 1.18.2.9.查询优化 ####

##### 1.18.2.9.1.Blink planner #####

Apache Flink使用并扩展了Apache Calcite来执行复杂的查询优化。 这包括一系列基于规则和成本的优化，例如：  
基于Apache Calcite的子查询解相关  
投影剪裁  
分区剪裁  
过滤器下推  
子计划消除重复数据以避免重复计算  
特殊子查询重写，包括两部分

1.  将 IN 和 EXISTS 转换为 left semi-joins
2.  将NOT IN 和 NOT EXISTS 转换为 left anti-join  
    可选join重新排序
3.  通过 table.optimizer.join-reorder-enabled 启用

\*\*注意：\*\*当前仅在子查询重写的结合条件下支持 IN / EXISTS / NOT IN / NOT EXISTS。

优化器不仅基于计划，而且还基于可从数据源获得的丰富统计信息以及每个算子（例如 io，cpu，网络和内存）的细粒度成本来做出明智的决策。

高级用户可以通过 CalciteConfig 对象提供自定义优化，可以通过调用 TableEnvironment＃getConfig＃setPlannerConfig 将其提供给 TableEnvironment。

##### 1.18.2.9.2.Old planner #####

Apache Flink利用Apache Calcite 来优化和翻译查询。当前执行的优化包括投影和过滤器下推，子查询消除以及其他类型的查询重写。原版计划程序尚未优化join 的顺序，而是按照查询中定义的顺序执行它们（FROM 子句中的表顺序和/或 WHERE 子句中的 join 谓词顺序）。

通过提供一个 CalciteConfig 对象，可以调整在不同阶段应用的优化规则集合。这个对象可以通过调用构造器 CalciteConfig.createBuilder() 创建，并通过调用tableEnv.getConfig.setPlannerConfig(calciteConfig) 提供给 TableEnvironment。

#### 1.18.2.10.解释表 ####

Table API 提供了一种机制来解释计算 Table 的逻辑和优化查询计划。 这是通过 Table.explain() 方法或者 StatementSet.explain() 方法来完成的。Table.explain() 返回一个 Table 的计划。StatementSet.explain() 返回多 sink 计划的结果。它返回一个描述三种计划的字符串：

关系查询的抽象语法树（the Abstract Syntax Tree），即未优化的逻辑查询计划，  
优化的逻辑查询计划，以及  
物理执行计划。  
可以用 TableEnvironment.explainSql() 方法和 TableEnvironment.executeSql() 方法支持执行一个 EXPLAIN 语句获取逻辑和优化查询计划，请参阅 EXPLAIN 页面.

以下代码展示了一个示例以及对给定Table使用Table.explain()方法的相应输出：  
**Java代码：**

package com.toto.demo.sql;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    import static org.apache.flink.table.api.Expressions.$;
    
    public class Demo { 
    
        public static void main(String[] args) { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
    
            DataStream<Tuple2<Integer, String>> stream1 = env.fromElements(new Tuple2<>(1,"hello"));
            DataStream<Tuple2<Integer, String>> stream2 = env.fromElements(new Tuple2<>(1, "hello"));
    
            // explain Table API
            Table table1 = tEnv.fromDataStream(stream1, $("count"), $("word"));
            Table table2 = tEnv.fromDataStream(stream2, $("count"), $("word"));
            Table table = table1
                    .where($("word").like("F%"))
                    .unionAll(table2);
            System.out.println(table.explain());
    
        }
    
    }

**输出结果：**

== Abstract Syntax Tree ==
    LogicalUnion(all=[true])
    :- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
    :  +- LogicalTableScan(table=[[Unregistered_DataStream_1]])
    +- LogicalTableScan(table=[[Unregistered_DataStream_2]])
    
    == Optimized Logical Plan ==
    Union(all=[true], union=[count, word])
    :- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
    :  +- DataStreamScan(table=[[Unregistered_DataStream_1]], fields=[count, word])
    +- DataStreamScan(table=[[Unregistered_DataStream_2]], fields=[count, word])
    
    == Physical Execution Plan ==
    Stage 1 : Data Source
    	content : Source: Collection Source
    
    Stage 2 : Data Source
    	content : Source: Collection Source
    
    	Stage 3 : Operator
    		content : SourceConversion(table=[Unregistered_DataStream_1], fields=[count, word])
    		ship_strategy : FORWARD
    
    		Stage 4 : Operator
    			content : Calc(select=[count, word], where=[(word LIKE _UTF-16LE'F%')])
    			ship_strategy : FORWARD
    
    			Stage 5 : Operator
    				content : SourceConversion(table=[Unregistered_DataStream_2], fields=[count, word])
    				ship_strategy : FORWARD

**Scala代码：**

package com.toto.learn.sql
    
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
    
    object Demo { 
    
      def main(args: Array[String]): Unit = { 
    
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val tEnv = StreamTableEnvironment.create(env)
    
        val table1 = env.fromElements((1, "hello")).toTable(tEnv, $"count", $"word")
        val table2 = env.fromElements((1, "hello")).toTable(tEnv, $"count", $"word")
        val table = table1
          .where($"word".like("F%"))
          .unionAll(table2)
        println(table.explain())
        
      }
    
    }

以下代码展示了一个示例以及使用StatementSet.explain()的多sink计划的相应输出：

EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
    TableEnvironment tEnv = TableEnvironment.create(settings);
    
    final Schema schema = new Schema()
        .field("count", DataTypes.INT())
        .field("word", DataTypes.STRING());
    
    tEnv.connect(new FileSystem().path("/source/path1"))
        .withFormat(new Csv().deriveSchema())
        .withSchema(schema)
        .createTemporaryTable("MySource1");
    tEnv.connect(new FileSystem().path("/source/path2"))
        .withFormat(new Csv().deriveSchema())
        .withSchema(schema)
        .createTemporaryTable("MySource2");
    tEnv.connect(new FileSystem().path("/sink/path1"))
        .withFormat(new Csv().deriveSchema())
        .withSchema(schema)
        .createTemporaryTable("MySink1");
    tEnv.connect(new FileSystem().path("/sink/path2"))
        .withFormat(new Csv().deriveSchema())
        .withSchema(schema)
        .createTemporaryTable("MySink2");
    
    StatementSet stmtSet = tEnv.createStatementSet();
    
    Table table1 = tEnv.from("MySource1").where($("word").like("F%"));
    stmtSet.addInsert("MySink1", table1);
    
    Table table2 = table1.unionAll(tEnv.from("MySource2"));
    stmtSet.addInsert("MySink2", table2);
    
    String explanation = stmtSet.explain();
    System.out.println(explanation);

**Scala代码：**

val settings = EnvironmentSettings.newInstance.useBlinkPlanner.inStreamingMode.build
    val tEnv = TableEnvironment.create(settings)
    
    val schema = new Schema()
        .field("count", DataTypes.INT())
        .field("word", DataTypes.STRING())
    
    tEnv.connect(new FileSystem().path("/source/path1"))
        .withFormat(new Csv().deriveSchema())
        .withSchema(schema)
        .createTemporaryTable("MySource1")
    tEnv.connect(new FileSystem().path("/source/path2"))
        .withFormat(new Csv().deriveSchema())
        .withSchema(schema)
        .createTemporaryTable("MySource2")
    tEnv.connect(new FileSystem().path("/sink/path1"))
        .withFormat(new Csv().deriveSchema())
        .withSchema(schema)
        .createTemporaryTable("MySink1")
    tEnv.connect(new FileSystem().path("/sink/path2"))
        .withFormat(new Csv().deriveSchema())
        .withSchema(schema)
        .createTemporaryTable("MySink2")
    
    val stmtSet = tEnv.createStatementSet()
    
    val table1 = tEnv.from("MySource1").where($"word".like("F%"))
    stmtSet.addInsert("MySink1", table1)
    
    val table2 = table1.unionAll(tEnv.from("MySource2"))
    stmtSet.addInsert("MySink2", table2)
    
    val explanation = stmtSet.explain()
    println(explanation)