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的子查詢解相關
?投影剪裁
?分區剪裁
?過濾器下推
?子計劃消除重復數據以避免重復計算
?特殊子查詢重寫，包括兩部分

將 IN 和 EXISTS 轉換為 left semi-joins

將NOT IN 和 NOT EXISTS 轉換為 left anti-join
?可選join重新排序

通過 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 APITable 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 Sourcecontent : Source: Collection SourceStage 2 : Data Sourcecontent : Source: Collection SourceStage 3 : Operatorcontent : SourceConversion(table=[Unregistered_DataStream_1], fields=[count, word])ship_strategy : FORWARDStage 4 : Operatorcontent : Calc(select=[count, word], where=[(word LIKE _UTF-16LE'F%')])ship_strategy : FORWARDStage 5 : Operatorcontent : SourceConversion(table=[Unregistered_DataStream_2], fields=[count, word])ship_strategy : FORWARD

Scala代碼：

package com.toto.learn.sqlimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.table.api.bridge.scala.StreamTableEnvironmentobject Demo {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentval 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)

總結

以上是生活随笔為你收集整理的1.18.2.9.查询优化、Blink planner、解释表的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。