1.18.2.10.解釋表

Table API 提供了一種機制來解釋計算 Table 的邏輯和優(yōu)化查詢計劃。 這是通過 Table.explain() 方法或者 StatementSet.explain() 方法來完成的。Table.explain() 返回一個 Table 的計劃。StatementSet.explain() 返回多 sink 計劃的結(jié)果。它返回一個描述三種計劃的字符串：

?關(guān)系查詢的抽象語法樹（the Abstract Syntax Tree），即未優(yōu)化的邏輯查詢計劃
?優(yōu)化的邏輯查詢計劃，以及
?物理執(zhí)行計劃。
可以用 TableEnvironment.explainSql() 方法和 TableEnvironment.executeSql() 方法支持執(zhí)行一個 EXPLAIN 語句獲取邏輯和優(yōu)化查詢計劃，請參閱 EXPLAIN 頁面.

以下代碼展示了一個示例以及對給定Table使用Table.explain()方法的相應(yīng)輸出：
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());}}

輸出結(jié)果：

== 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計劃的相應(yīng)輸出：

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)

總結(jié)

以上是生活随笔為你收集整理的1.18.2.10 解释表：Table.explain、物理执行计划等的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。