Apache Flink 有两种关系型 API 来做流批统一处理：Table API 和 SQL
- Table API 是用于 Scala 和 Java 语言的查询API，它可以用一种非常直观的方式来组合使用选取、过滤、join 等关系型算子。
- Flink SQL 是基于 Apache Calcite 来实现的标准 SQL。无论输入是连续的（流式）还是有界的（批处理），在两个接口中指定的查询都具有相同的语义，并指定相同的结果。
基本程序结构
import org.apache.flink.table.api.*;
import org.apache.flink.connector.datagen.table.DataGenOptions;// 可以从流中创建表
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//读取文件
DataStream<String> inputStream = env.readTextFile("xxx文件地址")//包装转换
DataStream<SensorReading> dataStream = xx转换;//创建表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//tableEnv
Table dataTable = tableEnv.fromDataSource(dataStream);Table resultTable = dataTable.select("id, temp")
.where("id=1");// 输出数据
tableEnv.toAppendStream(resultTable, Sensor.class);// register the Table dataTable as table "dataTable"
tableEnv.createTemporaryView("dataTable", dataTable);// 执行sql
Table resultTable1 = tableEnv.sqlQuery("select id,temp from dataTable where is = 1");// 输出数据
tableEnv.toAppendStream(resultTable1, Sensor.class);
TableEnvironment
TableAPI的核心概念是TableEnvironment，它主要负责
- 在内部的 catalog 中注册
Table - 注册外部的 catalog
- 加载可插拔模块
- 执行 SQL 查询
- 注册自定义函数 （scalar、table 或 aggregation）
DataStream和Table之间的转换(面向StreamTableEnvironment)
Table 总是与特定的 TableEnvironment 绑定。 不能在同一条查询中使用不同 TableEnvironment 中的表，例如，对它们进行 join 或 union 操作。 TableEnvironment 可以通过静态方法 TableEnvironment.create() 创建。
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode()//.inBatchMode().build();TableEnvironment tEnv = TableEnvironment.create(settings);或者也可以按程序示例中的方式，从现有的 StreamExecutionEnvironment 创建一个 StreamTableEnvironment 与 DataStream API 互操作。
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 env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);在 Catalog 中创建表
---先解释一下什么是Catalog：数据处理最关键的方面之一是管理元数据。 元数据可以是临时的，例如临时表、或者通过 TableEnvironment 注册的 UDF。 元数据也可以是持久化的，例如 Hive Metastore 中的元数据。Catalog 提供了一个统一的API，用于管理元数据，并使其可以从 Table API 和 SQL 查询语句中来访问。cataLog详细解释
TableEnvironment 维护着一个由标识符（identifier）创建的表 catalog 的映射。标识符由三个部分组成：catalog 名称、数据库名称以及对象名称。如果 catalog 或者数据库没有指明，就会使用当前默认值
TableEnvironment tEnv = ...;
tEnv.useCatalog("custom_catalog");
tEnv.useDatabase("custom_database");创建表
在 SQL 的术语中，Table API 的对象对应于视图（虚拟表）。它封装了一个逻辑查询计划。它可以通过以下方法在 catalog 中创建：
// register the Table dataTable as table "dataTable"
tableEnv.createTemporaryView("dataTable", dataTable);// 执行sql
Table resultTable = tableEnv.sqlQuery("select id,temp from dataTable where is = 1");// 输出数据
tableEnv.toAppendStream(resultTable, Sensor.class);注意： 从传统数据库系统的角度来看，Table 对象与 VIEW 视图非常像。也就是，定义了 Table 的查询是没有被优化的， 而且会被内嵌到另一个引用了这个注册了的 Table的查询中。如果多个查询都引用了同一个注册了的Table，那么它会被内嵌每个查询中并被执行多次， 也就是说注册了的Table的结果不会被共享。
Connector Tables
另外一个方式去创建 TABLE 是通过 connector 声明。Connector 描述了存储表数据的外部系统。存储系统例如 Apache Kafka 或者常规的文件系统都可以通过这种方式来声明。
// Using table descriptors
final TableDescriptor sourceDescriptor = TableDescriptor.forConnector("datagen").schema(Schema.newBuilder().column("f0", DataTypes.STRING()).build()).option(DataGenOptions.ROWS_PER_SECOND, 100).build();tableEnv.createTable("SourceTableA", sourceDescriptor);
tableEnv.createTemporaryTable("SourceTableB", sourceDescriptor);// Using SQL DDL
tableEnv.executeSql("CREATE [TEMPORARY] TABLE MyTable (...) WITH (...)")扩展表标识符
表总是通过三元标识符注册，包括 catalog 名、数据库名和表名。
用户可以指定一个 catalog 和数据库作为 “当前catalog” 和"当前数据库"。有了这些，那么刚刚提到的三元标识符的前两个部分就可以被省略了。如果前两部分的标识符没有指定， 那么会使用当前的 catalog 和当前数据库。用户也可以通过 Table API 或 SQL 切换当前的 catalog 和当前的数据库。
标识符遵循 SQL 标准，因此使用时需要用反引号（`）进行转义。
// 默认
TableEnvironment tEnv = ...;
tEnv.useCatalog("custom_catalog");
tEnv.useDatabase("custom_database");// 注册一个view 命名为 'exampleView' 在 catalog 中命名为 'custom_catalog'
// 默认在 'custom_database'的数据库中
tableEnv.createTemporaryView("exampleView", table);//注册一个view 命名为 'exampleView' 在 catalog 中命名为 'custom_catalog'
// 指定数据库的名称为 'other_database'
tableEnv.createTemporaryView("other_database.exampleView", table);// 注册一个view 命名为 'example.View'， 在 catalog 中命名为'custom_catalog'
// 默认在 'custom_database'的数据库中
tableEnv.createTemporaryView("`example.View`", table);// 注册view名为 'exampleView'， 指定 catalog命名为'other_catalog'
// 指定数据库的名称为 'other_database'
tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table);查询表
tableApi
Table API 是基于 Table 类的，该类表示一个表（流或批处理），并提供使用关系操作的方法。这些方法返回一个新的 Table 对象，该对象表示对输入 Table 进行关系操作的结果。 一些关系操作由多个方法调用组成，例如 table.groupBy(...).select()，其中 groupBy(...) 指定 table 的分组，而 select(...) 在 table 分组上的投影。
所有流处理和批处理表支持的 Table API 算子--> 文档 Table API
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section// 注册表// 获取order表
Table orders = tableEnv.from("Orders");
// 从所有的数据中过滤 cCountry=France，然后聚合
Table revenue = orders.filter($("cCountry").isEqual("FRANCE")).groupBy($("cID"), $("cName")).select($("cID"), $("cName"), $("revenue").sum().as("revSum"));// emit or convert Table
// execute querySQL
Flink SQL 是基于实现了SQL标准的 Apache Calcite 的。SQL 查询由常规字符串指定。
Flink对流处理和批处理表的SQL支持-->文档 SQL
一个示例：
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section// 注册 Orders table// 执行sql处理
Table revenue = tableEnv.sqlQuery("SELECT cID, cName, SUM(revenue) AS revSum " +"FROM Orders " +"WHERE cCountry = 'FRANCE' " +"GROUP BY cID, cName");// 展示了如何指定一个更新查询，将查询的结果插入到已注册的表中。
// 执行sql获取到的数据 and 提交到 "RevenueFrance"表中
tableEnv.executeSql("INSERT INTO RevenueFrance " +"SELECT cID, cName, SUM(revenue) AS revSum " +"FROM Orders " +"WHERE cCountry = 'FRANCE' " +"GROUP BY cID, cName");输出表
Table 通过写入 TableSink 输出。TableSink 是一个通用接口，用于支持多种文件格式（如 CSV、Apache Parquet、Apache Avro）、存储系统（如 JDBC、Apache HBase、Apache Cassandra、Elasticsearch）或消息队列系统（如 Apache Kafka、RabbitMQ）。
批处理
Table只能写入BatchTableSink，而流处理Table需要指定写入AppendStreamTableSink，RetractStreamTableSink或者UpsertStreamTableSink。
演示如何输出 Table：
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section// create an output Table
final Schema schema = Schema.newBuilder().column("a", DataTypes.INT()).column("b", DataTypes.STRING()).column("c", DataTypes.BIGINT()).build();tableEnv.createTemporaryTable("CsvSinkTable", TableDescriptor.forConnector("filesystem").schema(schema).option("path", "/path/to/file").format(FormatDescriptor.forFormat("csv").option("field-delimiter", "|").build()).build());// compute a result Table using Table API operators and/or SQL queries
Table result = ...//方法 Table.executeInsert(String tableName) 将 Table 发送至已注册的 TableSink。该方法通过名称在 catalog 中查找 TableSink 并确认Table schema 和 TableSink schema 一致。
result.executeInsert("CsvSinkTable");翻译与执行查询
不论输入数据源是流式的还是批式的，Table API 和 SQL 查询都会被转换成 DataStream 程序。 查询在内部表示为逻辑查询计划，并被翻译成两个阶段：
- 优化逻辑执行计划
- 翻译成 DataStream 程序
Table API 或者 SQL 查询在下列情况下会被翻译：
- 当
T ableEnvironment.executeSql()被调用时。该方法是用来执行一个 SQL 语句，一旦该方法被调用， SQL 语句立即被翻译。- 当
Table.executeInsert()被调用时。该方法是用来将一个表的内容插入到目标表中，一旦该方法被调用， TABLE API 程序立即被翻译。- 当
Table.execute()被调用时。该方法是用来将一个表的内容收集到本地，一旦该方法被调用， TABLE API 程序立即被翻译。- 当
StatementSet.execute()被调用时。Table（通过StatementSet.addInsert()输出给某个Sink）和 INSERT 语句 （通过调用StatementSet.addInsertSql()）会先被缓存到StatementSet中，StatementSet.execute()方法被调用时，所有的 sink 会被优化成一张有向无环图。- 当
Table被转换成DataStream时（参阅与 DataStream 集成）。转换完成后，它就成为一个普通的 DataStream 程序，并会在调用StreamExecutionEn vironment.execute()时被执行。
查询优化
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，网络和内存）的细粒度成本来做出明智的决策。
解释表
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() 方法的相应输出：
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 Physical 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])== Optimized Execution 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])参考资料：https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/table/common/#expanding-table-identifiers
