1.18.2.Table APISQL(概念与通用API、两种计划器（Planner）的主要区别、创建 TableEnvironment、临时表、永久表、创建表、虚拟表、Connector 等)

1.18.2.概念与通用API
1.18.2.1.两种计划器（Planner）的主要区别：
1.18.2.2.Table API和SQL程序的结构
1.18.2.3.创建 TableEnvironment
1.18.2.4.在Catalog中创建表
1.18.2.4.1.临时表（Temporary Table）和永久表（Permanent Table）
1.18.2.4.1.1.屏蔽（Shadowing）
1.18.2.4.2.创建表
1.18.2.4.2.1.虚拟表
1.18.2.4.2.2.Connector Tables
1.18.2.4.3.扩展表标识符

1.18.2.概念与通用API

Table API 和 SQL 集成在同一套 API 中。这套 API 的核心概念是Table，用作查询的输入和输出。本文介绍了 Table API 和 SQL 查询程序的通用结构、如何注册 Table 、如何查询 Table 以及如何输出 Table。

1.18.2.1.两种计划器（Planner）的主要区别：

1.Blink将批处理作业视作流处理的一种特例。严格来说，Table 和 DataSet 之间不支持相互转换，并且批处理作业也不会转换成 DataSet 程序而是转换成 DataStream 程序，流处理作业也一样。
2.Blink 计划器不支持 BatchTableSource，而是使用有界的 StreamTableSource 来替代。
3.旧计划器和 Blink 计划器中 FilterableTableSource 的实现是不兼容的。旧计划器会将 PlannerExpression 下推至 FilterableTableSource，而 Blink 计划器则是将 Expression 下推。
4.基于字符串的键值配置选项仅在 Blink 计划器中使用。（详情参见配置）
5.PlannerConfig 在两种计划器中的实现（CalciteConfig）是不同的。
6.Blink 计划器会将多sink（multiple-sinks）优化成一张有向无环图（DAG），TableEnvironment 和 StreamTableEnvironment 都支持该特性。旧计划器总是将每个sink都优化成一个新的有向无环图，且所有图相互独立。
7.旧计划器目前不支持 catalog 统计数据，而 Blink 支持。

1.18.2.2.Table API和SQL程序的结构

所有用于批处理和流处理的 Table API 和 SQL 程序都遵循相同的模式。下面的代码示例展示了 Table API 和 SQL 程序的通用结构。

其中pom.xml添加的依赖如下：

<!-- 取决于你使用的编程语言，选择Java或者Scala API来构建你的Table API和SQL程序 -->
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java-bridge_2.11</artifactId><version>1.12.0</version><!--<scope>provided</scope>-->
</dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-scala-bridge_2.11</artifactId><version>1.12.0</version><!--<scope>provided</scope>-->
</dependency><!-- 如果你想在 IDE 本地运行你的程序，你需要添加下面的模块，具体用哪个取决于你使用哪个 Planner -->
<!-- Either... (for the old planner that was available before Flink 1.9) -->
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_2.11</artifactId><version>1.12.0</version><!--<scope>provided</scope>-->
</dependency><!-- or.. (for the new Blink planner) -->
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner-blink_2.11</artifactId><version>1.12.0</version><!--<scope>provided</scope>-->
</dependency><!--
内部实现上，部分 table 相关的代码是用 Scala 实现的。所以，下面的依赖也需要添加到你的程序里，不管是批式还是流式的程序：
-->
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.11</artifactId><version>1.12.0</version><!--<scope>provided</scope>-->
</dependency><!-- 如果你想实现自定义格式来解析Kafka数据，或者自定义函数，下面的依赖就足够了，编译出来的jar文件可以直接给SQL Client使用 -->
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-common</artifactId><version>1.12.0</version><!--<scope>provided</scope>-->
</dependency><build><plugins><!-- 编译插件 --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.6.0</version><configuration><source>${maven.compiler.source}</source><target>${maven.compiler.target}</target><encoding>UTF-8</encoding><compilerVersion>${maven.compiler.source}</compilerVersion><showDeprecation>true</showDeprecation><showWarnings>true</showWarnings></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-surefire-plugin</artifactId><version>2.12.4</version><configuration><skipTests>${maven.test.skip}</skipTests></configuration></plugin><plugin><groupId>org.apache.rat</groupId><artifactId>apache-rat-plugin</artifactId><version>0.12</version><configuration><excludes><exclude>README.md</exclude></excludes></configuration></plugin><plugin><artifactId>maven-checkstyle-plugin</artifactId><version>2.17</version><executions><execution><id>verify</id><phase>verify</phase><configuration><configLocation>style/rmq_checkstyle.xml</configLocation><encoding>UTF-8</encoding><consoleOutput>true</consoleOutput><failsOnError>true</failsOnError><includeTestSourceDirectory>false</includeTestSourceDirectory><includeTestResources>false</includeTestResources></configuration><goals><goal>check</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-javadoc-plugin</artifactId><version>2.10.4</version><configuration><aggregate>true</aggregate><reportOutputDirectory>javadocs</reportOutputDirectory><locale>en</locale></configuration></plugin><!-- scala编译插件 --><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.1.6</version><configuration><scalaCompatVersion>2.11</scalaCompatVersion><scalaVersion>2.11.12</scalaVersion><encoding>UTF-8</encoding></configuration><executions><execution><id>compile-scala</id><phase>compile</phase><goals><goal>add-source</goal><goal>compile</goal></goals></execution><execution><id>test-compile-scala</id><phase>test-compile</phase><goals><goal>add-source</goal><goal>testCompile</goal></goals></execution></executions></plugin><!-- 打jar包插件(会包含所有依赖) --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>2.6</version><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs><archive><manifest><!-- 可以设置jar包的入口类(可选) --><mainClass></mainClass></manifest></archive></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins>
</build>

Java代码如下：

package com.toto.demo.sql;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;/*** @author tuzuoquan** @description** TableAPIAndSQLStructure** @date 2021/1/28 10:27**/
public class TableAPIAndSQLStructure {public static void main(String[] args) {//Flink STREAMING QUERYEnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();// create a TableEnvironment for specific planner batch or streaming//StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);//或者使用下面的方式TableEnvironment tableEnv = TableEnvironment.create(fsSettings);// create an input TabletableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )");// register an output TabletableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )");// create a Table object from a Table API queryTable table2 = tableEnv.from("table1").select(...);// create a Table object from a SQL queryTable table3 = tableEnv.sqlQuery("SELECT ... FROM table1 ... ");// emit a Table API result Table to a TableSink, same for SQL resultTableResult tableResult = table2.executeInsert("outputTable");tableResult...}}

scala代码如下：

package com.toto.learn.sqlimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}object TableAPIAndSQLStructure {def main(args: Array[String]): Unit = {//FLINK STREAMING QUERYval fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()val fsEnv = StreamExecutionEnvironment.getExecutionEnvironment//val fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings)// 或使用下面的创建方式val tableEnv = TableEnvironment.create(fsSettings)// create a TableEnvironment for specific planner batch or streaming//val tableEnv = ... // see "Create a TableEnvironment" section// create an input TabletableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )")// register an output TabletableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )")// create a Table from a Table API queryval table2 = tableEnv.from("table1").select(...)// create a Table from a SQL queryval table3 = tableEnv.sqlQuery("SELECT ... FROM table1 ...")// emit a Table API result Table to a TableSink, same for SQL resultval tableResult = table2.executeInsert("outputTable")tableResult...}}

1.18.2.3.创建 TableEnvironment

TableEnvironment是Table API和SQL的核心概念。它负责：
1.在内部的catalog中注册Table
2.注册外部的catalog
3.加载可插拔模块
4.执行SQL查询
5.注册自定义函数（scalar、table 或 aggregation）
6.将 DataStream 或 DataSet 转换成 Table
7.持有对 ExecutionEnvironment 或 StreamExecutionEnvironment 的引用。

Table 总是与特定的 TableEnvironment 绑定。不能在同一条查询中使用不同 TableEnvironment 中的表，例如，对它们进行 join 或 union 操作。

TableEnvironment 可以通过静态方法 BatchTableEnvironment.create() 或者 StreamTableEnvironment.create() 在 StreamExecutionEnvironment 或者 ExecutionEnvironment 中创建，TableConfig 是可选项。TableConfig可用于配置TableEnvironment或定制的查询优化和转换过程(参见查询优化（https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/common.html#query-optimization）)。

请确保选择与你的编程语言匹配的特定的计划器BatchTableEnvironment/StreamTableEnvironment。
如果两种计划器的 jar 包都在 classpath 中（默认行为），你应该明确地设置要在当前程序中使用的计划器。

JAVA版本

// **********************
// FLINK STREAMING QUERY
// **********************
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);
// or TableEnvironment fsTableEnv = TableEnvironment.create(fsSettings);// ******************
// FLINK BATCH QUERY
// ******************
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);// **********************
// BLINK STREAMING QUERY
// **********************
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
// or TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);// ******************
// BLINK BATCH QUERY
// ******************
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;EnvironmentSettings bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);

SCALA版本

// **********************
// FLINK STREAMING QUERY
// **********************
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironmentval fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
val fsEnv = StreamExecutionEnvironment.getExecutionEnvironment
val fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings)
// or val fsTableEnv = TableEnvironment.create(fsSettings)// ******************
// FLINK BATCH QUERY
// ******************
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.BatchTableEnvironmentval fbEnv = ExecutionEnvironment.getExecutionEnvironment
val fbTableEnv = BatchTableEnvironment.create(fbEnv)// **********************
// BLINK STREAMING QUERY
// **********************
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironmentval bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)
// or val bsTableEnv = TableEnvironment.create(bsSettings)// ******************
// BLINK BATCH QUERY
// ******************
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
val bbTableEnv = TableEnvironment.create(bbSettings)

注意：如果/lib目录中只有一种计划器的 jar 包，则可以使用useAnyPlanner（python 使用 use any_u_planner）创建 EnvironmentSettings。

1.18.2.4.在Catalog中创建表

TableEnvironment 维护着一个由标识符（identifier）创建的表 catalog 的映射。标识符由三个部分组成：catalog 名称、数据库名称以及对象名称。如果 catalog 或者数据库没有指明，就会使用当前默认值。（参见表标识符扩展章节中的例子）。

Table 可以是虚拟的（视图 VIEWS）也可以是常规的（表 TABLES）。视图VIEWS可以从已经存在的Table中创建，一般是Table API或者SQL的查询结果。表TABLES描述的是外部数据，例如文件、数据库表或者消息队列。

1.18.2.4.1.临时表（Temporary Table）和永久表（Permanent Table）

表可以是临时的，并与单个 Flink 会话（session）的生命周期相关，也可以是永久的，并且在多个 Flink 会话和群集（cluster）中可见。

永久表需要 catalog（例如 Hive Metastore）以维护表的元数据。一旦永久表被创建，它将对任何连接到 catalog 的 Flink 会话可见且持续存在，直至被明确删除。

1.18.2.4.1.1.屏蔽（Shadowing）

可以使用与已存在的永久表相同的标识符去注册临时表。临时表会屏蔽永久表，并且只要临时表存在，永久表就无法访问。所有使用该标识符的查询都将作用于临时表。

这可能对实验（experimentation）有用。它允许先对一个临时表进行完全相同的查询，例如只有一个子集的数据，或者数据是不确定的。一旦验证了查询的正确性，就可以对实际的生产表进行查询。

1.18.2.4.2.创建表

1.18.2.4.2.1.虚拟表

在 SQL 的术语中，Table API 的对象对应于视图（虚拟表）。它封装了一个逻辑查询计划。它可以通过以下方法在 catalog 中创建：

package com.toto.demo.sql;import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;/*** @author tuzuoquan* @version 1.0* @ClassName CreateTable* @description TODO* @date 2021/1/29 10:24**/
public class CreateTable {public static void main(String[] args) {EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();// get a TableEnvironmentTableEnvironment tableEnv = TableEnvironment.create(bsSettings);// table is the result of a simple projection queryTable projTable = tableEnv.from("X").select(...);// register the Table projTable as table "projectedTable"tableEnv.createTemporaryView("projectedTable", projTable);}}

Scala代码

package com.toto.learn.sqlimport org.apache.flink.table.api.{EnvironmentSettings, Table, TableEnvironment}object CreateTable {def main(args: Array[String]): Unit = {val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()val tableEnv = TableEnvironment.create(bbSettings)// table is the result of a simple projection query val projTable: Table = tableEnv.from("X").select(...)tableEnv.createTemporaryView("projectedTable",projTable)}}

注意：从传统数据库系统的角度来看，Table对象与VIEW视图非常像。也就是，定义了Table的查询是没有被优化的，而且会被内嵌到另一个引用了这个注册了的Table的查询中。如果多个查询都引用了同一个注册了的Table，那么它会被内嵌每个查询中被执行多次，也就是说注册了的Table的结果不会被共享（注：Blink计划器的TableEnvironment会被优化成只执行了一次）。

1.18.2.4.2.2.Connector Tables

另外一个方式去创建 TABLE 是通过 connector 声明。Connector 描述了存储表数据的外部系统。存储系统例如 Apache Kafka 或者常规的文件系统都可以通过这种方式来声明。

java代码：

tableEnvironment.connect(...).withFormat(...).withSchema(...).inAppendMode().createTemporaryTable("MyTable")

Scala代码：

tableEnvironment.connect(...).withFormat(...).withSchema(...).inAppendMode().createTemporaryTable("MyTable")

1.18.2.4.3.扩展表标识符

表总是通过三元标识符注册，包括 catalog 名、数据库名和表名。

用户可以指定一个 catalog 和数据库作为 “当前catalog” 和”当前数据库”。有了这些，那么刚刚提到的三元标识符的前两个部分就可以被省略了。如果前两部分的标识符没有指定，那么会使用当前的 catalog 和当前数据库。用户也可以通过 Table API 或 SQL 切换当前的 catalog 和当前的数据库。

标识符遵循 SQL 标准，因此使用时需要用反引号（`）进行转义。
Java代码

package com.toto.demo.sql;import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;/*** @author tuzuoquan* @version 1.0* @ClassName * @description TODO* @date 2021/1/29 10:24**/
public class Demo {public static void main(String[] args) {EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();// get a TableEnvironmentTableEnvironment tableEnv = TableEnvironment.create(bsSettings);tableEnv.useCatalog("custom_catalog");tableEnv.useDatabase("custom_database");Table table = tableEnv.sqlQuery("SELECT ... FROM table1 ... ");// register the view named 'exampleView' in the catalog named 'custom_catalog'// in the database named 'custom_database'tableEnv.createTemporaryView("exampleView", table);// register the view named 'exampleView' in the catalog named 'custom_catalog'// in the database named 'other_database'tableEnv.createTemporaryView("other_database.exampleView", table);// register the view named 'example.View' in the catalog named 'custom_catalog'// in the database named 'custom_database'tableEnv.createTemporaryView("`example.View`", table);// register the view named 'exampleView' in the catalog named 'other_catalog'// in the database named 'other_database'tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table);}}

Scala代码

package com.toto.learn.sqlimport org.apache.flink.table.api.{EnvironmentSettings, Table, TableEnvironment}object Demo {def main(args: Array[String]): Unit = {val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()val tableEnv: TableEnvironment = TableEnvironment.create(bbSettings)tableEnv.useCatalog("custom_catalog")tableEnv.useDatabase("custom_database")val table: Table = tableEnv.sqlQuery("SELECT ... FROM table1 ...")// register the view named 'exampleView' in the catalog named 'custom_catalog'// in the database named 'custom_database'tableEnv.createTemporaryView("exampleView", table)// register the view named 'exampleView' in the catalog named 'custom_catalog'// in the database named 'other_database'tableEnv.createTemporaryView("other_database.exampleView", table)// register the view named 'example.View' in the catalog named 'custom_catalog'// in the database named 'custom_database'tableEnv.createTemporaryView("`example.View`", table)// register the view named 'exampleView' in the catalog named 'other_catalog'// in the database named 'other_database'tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table)}}