1.18.2.10.解釋表

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

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

以下代碼展示了一個(gè)示例以及對(duì)給定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())}}

以下代碼展示了一個(gè)示例以及使用StatementSet.explain()的多sink計(jì)劃的相應(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、物理执行计划等的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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