2、Flink Table API & SQL - 概念和通用API

小鱼儿 2024-04-20 09:30 63阅读 0赞

Table API 和 SQL 集成在一个 API 中。这个 API 用作查询、输入和输出的表。本文档展示了带有 Table API 和 SQL 查询的程序的公共结构、如何注册表、如何查询表以及如何写入表。

**目录**

两个Planner之间的主要区别

Table API and SQL的结构

创建一个TableEnvironment

在 Calalog 中注册表

注册外部Calalog

查询表

Table API

SQL

混合使用 Table API & SQL

写表

转化并执行查询

Old Planner

Blink Planner

与DataStream和DataSet API集成

Scala的隐式转换

将DataStream和DataSet注册为表

将DataStream和DataSet转换为表

将表转换为DataStream或DataSet

将表转换为DataStream

将表转换为DataSet

数据类型到表模式的映射

基于位置映射

基于名称映射

原子类型

元组(Scala和Java)和Case类(仅限Scala)

POJO (Java and Scala)

Row

优化查询

解释表

--------------------

# 两个Planner之间的主要区别 #

1.  Blink 将批作业看作一个特殊的流。因此，Table 和 DataSet 之间不能转换，批处理作业不会被转换为 DateSet 程序，而是转换为 DataStream 程序，与流作业相同。
2.  Blink planner 不支持 `BatchTableSource，所以要使用 StreamTableSource。`
3.  Blink planner 支持新的 Catalog，不支持 `ExternalCatalog。`
4.  旧 planner 和 Blink planner 的 `FilterableTableSource 实现是互不兼容的，旧的把 PlannerExpression`s 下发给 `FilterableTableSource，Blink planner 下发给 Expression`s。
5.  基于字符串的键值配置选项仅用于 Blink planner，详情参阅：//TODO。
6.  两个 planner 中 PlannerConfig 的实现(CalciteConfig)是不同的。
7.  Blink planner 将优化多个 sink 到一个 DAG 中（只支持 `TableEnvironment`）。旧 planner 一个 sink 对应一个新的 DAG，每个 DAG 都是相对独立的。
8.  旧 planner 不支持 catalog，Blink planner支持。

# Table API and SQL的结构 #

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

// create a TableEnvironment for specific planner batch or streaming
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // register a Table
    tableEnv.registerTable("table1", ...)           // or
    tableEnv.registerTableSource("table2", ...)     // or
    tableEnv.registerExternalCatalog("extCat", ...)
    // register an output Table
    tableEnv.registerTableSink("outputTable", ...);
    
    // create a Table from a Table API query
    val tapiResult = tableEnv.scan("table1").select(...)
    // create a Table from a SQL query
    val sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table2 ...")
    
    // emit a Table API result Table to a TableSink, same for SQL result
    tapiResult.insertInto("outputTable")
    
    // execute
    tableEnv.execute("scala_job")

> **注意：**Table API & SQL 查询可以很容易地与 DataStream 或 DataSet 程序集成并嵌入其中。查阅下方：与DataStream和DataSet API集成，了解如何将数据流和数据集转换为表。

# 创建一个TableEnvironment #

TableEnvironment 是 Table API & SQL 集成的核心概念。负责:：

*  在内部 catalog 中注册表
 *  注册catalog
 *  执行 SQL 查询
 *  注册用户自定义函数（scalar, table, or aggregation）
 *  将 `DataStream` 或 `DataSet 转换为表`
 *  保存对 ExecutionEnvironment 或 StreamExecutionEnvironment 的引用

表总是绑定在特定的 `TableEnvironment 上，`在同一个查询中组合不同 TableEnvironment 的表是不可能的，例如，join 或 union 它们。

使用 StreamExecutionEnvironment 或 ExecutionEnvironment 和可选的 TableConfig 调用静态 BatchTableEnvironment.create()或 StreamTableEnvironment .create()方法创建 TableEnvironment。`TableConfig 可以用来配置 TableEnvironment 或 自定义优化查询（详情见下：查询优化）`

`BatchTableEnvironment`/`StreamTableEnvironment 指定匹配的 planner。`

如果两个 planner 的jar都存在，则需要在程序中显式指定 planner，如下：

// **********************
    // FLINK STREAMING QUERY
    // **********************
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.EnvironmentSettings
    import org.apache.flink.table.api.scala.StreamTableEnvironment
    
    val 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.scala.BatchTableEnvironment
    
    val 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.scala.StreamTableEnvironment
    
    val 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目录中只有一个planner jar，那么可以使用 useAnyPlanner(use\_any\_planner for python)创建特定的 `EnvironmentSettings`。

# 在 Calalog 中注册表 #

TableEnvironment 维护按名称注册的表的目录。有两种类型的表，输入表和输出表。输入表可以在 Table API & SQL 查询中引用，并提供输入数据。输出表可用于将 Table API & SQL 查询的结果输出到外部系统。

输入表的注册支持各种各样的数据源：

*  现有表对象，通常是 Table API & SQL 查询的结果。
 *  一个 TableSource，访问外部数据，例如一个文件，数据库，消息系统。
 *  DataStream（仅用于流任务） 和 DataSet（仅用于从旧 planner 转换的批作业） 程序中的 DataStream 和 DataSet。注册 DataStream 和 DataSet 查阅下方：与DataStream和DataSet API集成

使用 `TableSink 注册输入表。`

## 注册 Table ##

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // table is the result of a simple projection query 
    val projTable: Table = tableEnv.scan("X").select(...)
    
    // register the Table projTable as table "projectedTable"
    tableEnv.registerTable("projectedTable", projTable)

>  **注意：**注册表的处理类似于关系数据库系统中的视图，即，定义表的查询未进行优化，但当另一个查询引用已注册表时，将内联该查询。如果多个查询引用了相同注册表，它将为每个引用查询内联并执行多次，即，已注册表的结果将不会共享。

## 注册 TableSource ##

TableSource提供了访问外部数据，诸如，数据库（MySQL, HBase, …），特定编码的文件（CSV, Apache \[Parquet, Avro, ORC\], …），消息系统（Apache Kafka, RabbitMQ, …）。

Flink 旨在为常见的数据格式和存储系统提供 TableSources。查阅：//TODO，了解支持的 Table Source 列表及如何构建自定义 Table Source。

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // create a TableSource
    val csvSource: TableSource = new CsvTableSource("/path/to/file", ...)
    
    // register the TableSource as table "CsvTable"
    tableEnv.registerTableSource("CsvTable", csvSource)

> **注意：**用于 Blink planner 的 `TableEnvironment `只接受 StreamTableSource、LookupableTableSource 和 InputFormatTableSource，并且必须绑定用于批量 Blink planner 的流表资源。

## 注册 TableSink ##

已注册的 `TableSink `可以用于将 Table API & SQL 查询的结果输出到外部存储系统，诸如，数据库（MySQL, HBase, …），特定编码的文件（CSV, Apache \[Parquet, Avro, ORC\], …），消息系统（Apache Kafka, RabbitMQ, …）。

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // create a TableSink
    val csvSink: TableSink = new CsvTableSink("/path/to/file", ...)
    
    // define the field names and types
    val fieldNames: Array[String] = Array("a", "b", "c")
    val fieldTypes: Array[TypeInformation[_]] = Array(Types.INT, Types.STRING, Types.LONG)
    
    // register the TableSink as table "CsvSinkTable"
    tableEnv.registerTableSink("CsvSinkTable", fieldNames, fieldTypes, csvSink)

# 注册外部Calalog #

外部目录可以提供关于外部数据库和表的信息，比如它们的名称、模式、统计信息，以及如何访问存储在外部数据库、表或文件中的数据的信息。

外部目录可以通过实现 ExternalCatalog 接口创建，并在 TableEnvironment 中注册样例如下：

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // create an external catalog
    val catalog: ExternalCatalog = new InMemoryExternalCatalog
    
    // register the ExternalCatalog catalog
    tableEnv.registerExternalCatalog("InMemCatalog", catalog)

在 TableEnvironment 中注册后，可以通过指定 Table API 或 SQL 查询的完整路径（如：catalog.database.table）访问 ExternalCatalog 中定义的所有表。

目前，Flink 提供了一个用于演示和测试目的的 InMemoryExternalCatalog。然而，ExternalCatalog 接口还可以用于将 HCatalog 或 Metastore 之类的目录连接到 Table API。

> **注意：**Blink planner 不支持外部目录

# 查询表 #

## Table API ##

Table API 是一个用于Scala和Java的语言集成查询 API。与 SQL 相反，查询没有指定为sql，而是在程序中逐步拼接成。

Table API 基于 Table 类， 表示一个表(流或批处理)，并提供应用关系操作的方法。这些方法生成一个新的表对象，表示在输入表的基础上计算完成后的结果。一些计算由多个方法组成，如，`table.groupBy(...).select()。`

更多关于 Table API 的使用请查阅：//TODO

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // register Orders table
    
    // scan registered Orders table
    val orders = tableEnv.scan("Orders")
    // compute revenue for all customers from France
    val revenue = orders
      .filter('cCountry === "FRANCE")
      .groupBy('cID, 'cName)
      .select('cID, 'cName, 'revenue.sum AS 'revSum)
    
    // emit or convert Table
    // execute query

> **注意：**Scala Table API 使用一个单引号（'）来表示引用表。Table API使用的是 Scala 隐式，必须导入`org.apache.flink.api.scala._` 和 `org.apache.flink.table.api.scala._`

## SQL ##

Flink SQL 的集成基于 Apache Calcite，Calcite 实现了 SQL 标准，SQL 查询被指定为sql字符串。

下面的例子展示了如何指定一个查询，并返回结果数据表：

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // register Orders table
    
    // compute revenue for all customers from France
    val revenue = tableEnv.sqlQuery("""
      |SELECT cID, cName, SUM(revenue) AS revSum
      |FROM Orders
      |WHERE cCountry = 'FRANCE'
      |GROUP BY cID, cName
      """.stripMargin)
    
    // emit or convert Table
    // execute query

下面的示例显示如何指定将结果插入已注册表的 update 查询：

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // register "Orders" table
    // register "RevenueFrance" output table
    
    // compute revenue for all customers from France and emit to "RevenueFrance"
    tableEnv.sqlUpdate("""
      |INSERT INTO RevenueFrance
      |SELECT cID, cName, SUM(revenue) AS revSum
      |FROM Orders
      |WHERE cCountry = 'FRANCE'
      |GROUP BY cID, cName
      """.stripMargin)
    
    // execute query

## 混合使用 Table API & SQL ##

因为 Table API 和 SQL 返回的都是 Table 对象，所以他们是可以混合使用的。

*  可以在 SQL 查询返回的表对象上定义 Table API 查询。
 *  通过在 TableEnvironment 中注册结果表并在 SQL 查询的 FROM 子句中引用它，可以根据 Table API 查询的结果定义 SQL 查询。

# 写表 #

Table 是通过 TableSink 将数据写出的。TableSink 是一个通用的接口，支持多种文件格式（CSV, Apache Parquet, Apache Avro），存储系统（JDBC, Apache HBase, Apache Cassandra, Elasticsearch），消息系统（Apache Kafka, RabbitMQ）。

批处理表只能写入 BatchTableSink，而流表需要 AppendStreamTableSink、RetractStreamTableSink 或 UpsertStreamTableSink。

有关可用 Sink 的详细信息，以及如何实现自定义 TableSink 的说明，请参阅：//TODO。

`Table.insertInto(String tableName) 方法将一个 Table 写入到已经注册的 TableSink。该方法通过名称在 catalog 中查询 TableSink 并验证 Table Schema 和 TableSink Schema 是否一致。`

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // create a TableSink
    val sink: TableSink = new CsvTableSink("/path/to/file", fieldDelim = "|")
    
    // register the TableSink with a specific schema
    val fieldNames: Array[String] = Array("a", "b", "c")
    val fieldTypes: Array[TypeInformation] = Array(Types.INT, Types.STRING, Types.LONG)
    tableEnv.registerTableSink("CsvSinkTable", fieldNames, fieldTypes, sink)
    
    // compute a result Table using Table API operators and/or SQL queries
    val result: Table = ...
    
    // emit the result Table to the registered TableSink
    result.insertInto("CsvSinkTable")
    
    // execute the program

# 转化并执行查询 #

对于现有的两个 planner，转化和执行查询是不相同的。

## Old Planner ##

Table API 和 SQL 查询根据其输入是流输入还是批输入被转换为 DataStream 或 DataSet 程序。查询在内部表示为逻辑查询计划，并分两个阶段解释：

1.  优化逻辑计划
2.  转换为 DataStream 或 DataSet 程序

Table API 或 SQL 查询在以下情况下转化：

*  调用 `Table.insertInto()方法将 Table 写入 TableSink`
 *  调用 `TableEnvironment.sqlUpdate()方法指定 SQL 更新查询`
 *  Table 被转化成 DataStream 或 DataSet（查阅：与DataStream和DataSet API集成）

转换后，Table API 或 SQL 查询的处理方式与常规 DataStream 或 DataSet 程序类似，并在调用 StreamExecutionEnvironment.execute()或 ExecutionEnvironment.execute()时执行。

## Blink Planner ##

Table API 和 SQL 查询被转换为数据流程序或批处理程序。查询在内部表示为逻辑查询计划，并分两个阶段转换：

1.  优化逻辑计划
2.  转换为 DataStream 程序

对于 TableEnvironment 和 StreamTableEnvironment，转换查询是不同的。对于 TableEnvironment，当调用TableEnvironment.execute()时，将转换 Table API 或 SQL 查询，因为 TableEnvironment 将把多个 Sink 优化为一个DAG。

而对于 StreamTableEnvironment，当以下情况发生时，Table API 和 SQL 才会被转换:

*  调用 `Table.insertInto()方法将 Table 写入 TableSink`
 *  调用 `TableEnvironment.sqlUpdate()方法指定 SQL 更新查询`
 *  Table 转换成 DataStream

一旦转换完成，Table API 或 SQL 查询就像普通的 DataStream 程序一样处理，并在调用 TableEnvironment.execute() 或 StreamExecutionEnvironment.execute() 时执行。

# 与DataStream和DataSet API集成 #

流上的两个 planner 都可以与 DataStream API 集成。只有老的 planner 才能与 DataSet API 集成，批量上的 Blink planner 不能同时与两者结合。**注意：**下面讨论的 DataSet API 只适用于旧的批处理 planner。

Table API 和 SQL 查询可以很容易地与 DataStream 和 DataSet 程序集成并嵌入其中。例如，可以查询外部表（例如，RDBMS），做一些预处理，如 filtering, projecting, aggregating, joining 元数据，然后进一步处理 DataStream 数据或 DataSet 的API（任何构建在这些api之上的库，如 CEP 或 Gelly）。相反，Table API 或 SQL 查询也可以应用在 DataStream 或 DataSet 程序的结果上。

这种交互可以通过将 DataStream 或 DataSet 转换为表来实现，反之亦然。

## Scala的隐式转换 ##

Scala Table API 为 DataSet、DataStream 和 Table 类提供隐式转换。通过导入org.apache.flink.table.api.scala包，可以启用这些转换。除了org.apache.flink.api.scala。\_用于Scala DataStream API。

## 将DataStream和DataSet注册为表 ##

DataStream 或 DataSet 可以在 `TableEnvironment `中注册为表。结果表的模式取决于已注册的 DataStream 或 DataSet 的数据类型。有关数据类型到表模式的映射的详细信息，请参阅：数据类型到表模式的映射。

// get TableEnvironment 
    // registration of a DataSet is equivalent
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    val stream: DataStream[(Long, String)] = ...
    
    // register the DataStream as Table "myTable" with fields "f0", "f1"
    tableEnv.registerDataStream("myTable", stream)
    
    // register the DataStream as table "myTable2" with fields "myLong", "myString"
    tableEnv.registerDataStream("myTable2", stream, 'myLong, 'myString)

> **注意：**DataStream Table 名称不能与 `^_DataStreamTable_[0-9]+ 相匹配，DataSet Table 名称不能与 ^_DataSetTable_[0-9]+ 相匹配。因为他们仅提供给内部使用。`

## 将DataStream和DataSet转换为表 ##

不需要在 `TableEnvironment `中注册 DataStream 或 DataSet，它还可以直接转换为表。

// get TableEnvironment
    // registration of a DataSet is equivalent
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    val stream: DataStream[(Long, String)] = ...
    
    // convert the DataStream into a Table with default fields '_1, '_2
    val table1: Table = tableEnv.fromDataStream(stream)
    
    // convert the DataStream into a Table with fields 'myLong, 'myString
    val table2: Table = tableEnv.fromDataStream(stream, 'myLong, 'myString)

## 将表转换为DataStream或DataSet ##

表可以转换为 DataStream 或 DataSet。通过这种方式，可以对 Table API 或 SQL 查询的结果运行自定义 DataStream 或 DataSet 程序。

在将表转换为 DataStream 或 DataSet 时，需要指定结果 DataStream 或 DataSet 的数据类型，即，要将表的行转换成的数据类型。通常最方便的转换类型是 Row。以下是不同选项的功能概览：

*  **Row：**字段由位置、任意数量的字段、对空值的支持、没有类型安全访问来映射。
 *  **POJO：**字段按名称映射(POJO字段必须命名为表字段)、任意数量的字段、支持空值、类型安全访问。
 *  **Case Class：**字段按位置映射，不支持空值，类型安全访问。
 *  **Tuple：**字段按位置映射，限制为22 (Scala)或25 (Java)字段，不支持空值，类型安全访问。
 *  **Atomic Type：**表必须有一个字段，不支持空值，类型安全访问。

### 将表转换为DataStream ###

流查询的结果表将被动态更新，即，随着新记录到达查询的输入流，它也在发生变化。因此，将这样的动态查询转换成的 DataStream 需要对表的更新进行编码。

将表转换为 DataStream 有两种模式：

1.  **追加：**只有动态表被 insert 修改时，该模式才可用。它只追加数据，之前写入的数据不会被修改。
2.  **撤回：**这种模式总是可以使用。它使用 `boolean `标志对插入和删除更改进行编码。

// get TableEnvironment. 
    // registration of a DataSet is equivalent
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    // Table with two fields (String name, Integer age)
    val table: Table = ...
    
    // convert the Table into an append DataStream of Row
    val dsRow: DataStream[Row] = tableEnv.toAppendStream[Row](table)
    
    // convert the Table into an append DataStream of Tuple2[String, Int]
    val dsTuple: DataStream[(String, Int)] dsTuple = 
      tableEnv.toAppendStream[(String, Int)](table)
    
    // convert the Table into a retract DataStream of Row.
    //   A retract stream of type X is a DataStream[(Boolean, X)]. 
    //   The boolean field indicates the type of the change. 
    //   True is INSERT, false is DELETE.
    val retractStream: DataStream[(Boolean, Row)] = tableEnv.toRetractStream[Row](table)

> **注意：**关于动态表及其属性的详细讨论在动态表文档中给出（//TODO）。

### 将表转换为DataSet ###

将表转换为 DataSet 的过程如下：

// get TableEnvironment 
    // registration of a DataSet is equivalent
    val tableEnv = BatchTableEnvironment.create(env)
    
    // Table with two fields (String name, Integer age)
    val table: Table = ...
    
    // convert the Table into a DataSet of Row
    val dsRow: DataSet[Row] = tableEnv.toDataSet[Row](table)
    
    // convert the Table into a DataSet of Tuple2[String, Int]
    val dsTuple: DataSet[(String, Int)] = tableEnv.toDataSet[(String, Int)](table)

## 数据类型到表模式的映射 ##

Flink 的 DataStream 和 DataSet APIs 支持不同的类型。复合类型如元组(内置Scala和Flink Java元组)、pojo、Scala case类和Flink 的 Row 类型，允许嵌套数据结构，其中包含多个字段，可以在 table 表达式中访问。其他类型被视为原子类型。在下面的文章中，我们将描述 Table API 如何将这些类型转换为内部行表示，并展示将 DataStream 转换为表的示例。

数据类型到表模式的映射可以通过两种方式进行：**基于字段位置或基于字段名称**。

### 基于位置映射 ###

基于位置的映射可用于在保持字段顺序的同时赋予字段更有意义的名称。此映射可用于具有已定义字段顺序的组合数据类型和原子类型。元组、行和case类等组合数据类型具有这样的字段顺序。但是，必须根据字段名映射POJO的字段(请参阅下一节)。字段可以被投影出来，但不能使用别名重命名。

在定义基于位置的映射时，输入数据类型中**必须不存在指定的名称**，否则API将假定映射应该基于字段名称进行。如果没有指定字段名，则使用组合类型的默认字段名和字段顺序，或者使用f0表示原子类型。

// get a TableEnvironment
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    val stream: DataStream[(Long, Int)] = ...
    
    // convert DataStream into Table with default field names "_1" and "_2"
    val table: Table = tableEnv.fromDataStream(stream)
    
    // convert DataStream into Table with field "myLong" only
    val table: Table = tableEnv.fromDataStream(stream, 'myLong)
    
    // convert DataStream into Table with field names "myLong" and "myInt"
    val table: Table = tableEnv.fromDataStream(stream, 'myLong, 'myInt)

### 基于名称映射 ###

基于名称的映射可以用于任何数据类型，包括pojo。这是定义表模式映射最灵活的方法。映射中的所有字段都按名称引用，可以使用别名 as 重命名。字段可以重新排序并投影出来。

如果没有指定字段名，则使用组合类型的默认字段名和字段顺序，或者使用f0表示原子类型。

// get a TableEnvironment
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    val stream: DataStream[(Long, Int)] = ...
    
    // convert DataStream into Table with default field names "_1" and "_2"
    val table: Table = tableEnv.fromDataStream(stream)
    
    // convert DataStream into Table with field "_2" only
    val table: Table = tableEnv.fromDataStream(stream, '_2)
    
    // convert DataStream into Table with swapped fields
    val table: Table = tableEnv.fromDataStream(stream, '_2, '_1)
    
    // convert DataStream into Table with swapped fields and field names "myInt" and "myLong"
    val table: Table = tableEnv.fromDataStream(stream, '_2 as 'myInt, '_1 as 'myLong)

### 原子类型 ###

Flink 将基本类型(`Integer`, `Double`, `String`)或泛型类型(不能分析和分解的类型)视为原子类型。原子类型的 `DataStream `或 `DataStream `被转换为具有单个属性的表。属性的类型从原子类型推断，可以指定属性的名称。

// get a TableEnvironment
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    val stream: DataStream[Long] = ...
    
    // convert DataStream into Table with default field name "f0"
    val table: Table = tableEnv.fromDataStream(stream)
    
    // convert DataStream into Table with field name "myLong"
    val table: Table = tableEnv.fromDataStream(stream, 'myLong)

### 元组(Scala和Java)和Case类(仅限Scala) ###

Flink支持Scala的内置元组，并为Java提供了自己的元组类。这两种元组的数据流和数据集都可以转换为表。可以通过为所有字段提供名称(基于位置的映射)来重命名字段。如果没有指定字段名，则使用默认字段名。如果引用原始字段名(Flink元组的f0、f1、…和Scala元组的\_1、\_2、…)，则API假定映射是基于名称的，而不是基于位置的。基于名称的映射允许使用别名(as)对字段和投影进行重新排序。

// get a TableEnvironment
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    val stream: DataStream[(Long, String)] = ...
    
    // convert DataStream into Table with renamed default field names '_1, '_2
    val table: Table = tableEnv.fromDataStream(stream)
    
    // convert DataStream into Table with field names "myLong", "myString" (position-based)
    val table: Table = tableEnv.fromDataStream(stream, 'myLong, 'myString)
    
    // convert DataStream into Table with reordered fields "_2", "_1" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, '_2, '_1)
    
    // convert DataStream into Table with projected field "_2" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, '_2)
    
    // convert DataStream into Table with reordered and aliased fields "myString", "myLong" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, '_2 as 'myString, '_1 as 'myLong)
    
    // define case class
    case class Person(name: String, age: Int)
    val streamCC: DataStream[Person] = ...
    
    // convert DataStream into Table with default field names 'name, 'age
    val table = tableEnv.fromDataStream(streamCC)
    
    // convert DataStream into Table with field names 'myName, 'myAge (position-based)
    val table = tableEnv.fromDataStream(streamCC, 'myName, 'myAge)
    
    // convert DataStream into Table with reordered and aliased fields "myAge", "myName" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, 'age as 'myAge, 'name as 'myName)

### POJO (Java and Scala) ###

Flink支持将pojo作为复合类型。请查阅：[Flink基础之API，DataSet、DataStream、批、流][Flink_API_DataSet_DataStream]

在不指定字段名的情况下将POJO DataStream或DataSet转换为表时，将使用原始POJO字段的名称。名称映射需要原始名称，不能通过位置来完成。字段可以使用别名(带as关键字)重命名、重新排序和投影。

// get a TableEnvironment
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    // Person is a POJO with field names "name" and "age"
    val stream: DataStream[Person] = ...
    
    // convert DataStream into Table with default field names "age", "name" (fields are ordered by name!)
    val table: Table = tableEnv.fromDataStream(stream)
    
    // convert DataStream into Table with renamed fields "myAge", "myName" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, 'age as 'myAge, 'name as 'myName)
    
    // convert DataStream into Table with projected field "name" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, 'name)
    
    // convert DataStream into Table with projected and renamed field "myName" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, 'name as 'myName)

### Row ###

行数据类型支持任意数量的字段和具有空值的字段。字段名可以通过RowTypeInfo指定，也可以在将行DataStream或DataSet转换为表时指定。行类型支持按位置和名称映射字段。字段可以通过为所有字段提供名称(基于位置的映射)来重命名，或者单独选择投影/排序/重命名(基于名称的映射)。

// get a TableEnvironment
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    // DataStream of Row with two fields "name" and "age" specified in `RowTypeInfo`
    val stream: DataStream[Row] = ...
    
    // convert DataStream into Table with default field names "name", "age"
    val table: Table = tableEnv.fromDataStream(stream)
    
    // convert DataStream into Table with renamed field names "myName", "myAge" (position-based)
    val table: Table = tableEnv.fromDataStream(stream, 'myName, 'myAge)
    
    // convert DataStream into Table with renamed fields "myName", "myAge" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, 'name as 'myName, 'age as 'myAge)
    
    // convert DataStream into Table with projected field "name" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, 'name)
    
    // convert DataStream into Table with projected and renamed field "myName" (name-based)
    val table: Table = tableEnv.fromDataStream(stream, 'name as 'myName)

## 优化查询 ##

**Old planner**

Apache Flink 利用Apache Calcite 优化和解释查询。当前执行的优化包括 projection and filter push-down, subquery decorrelation, and other kinds of query rewriting。Old planner还没有优化 join 顺序，但是会按照查询中定义的顺序执行它们(FROM子句中的表顺序和/或WHERE子句中的连接谓词顺序)。

通过提供一个 CalciteConfig 对象，可以调整应用于不同阶段的优化规则集。这可以通过调用 calciteConfig . createbuilder()通过构建器创建，并通过调用 tableEnv.getConfig.setPlannerConfig(calciteConfig)提供给 TableEnvironment。

**Blink Planner**

Apache Flink 利用并扩展了 Apache Calcite 来执行复杂的查询优化。这包括一系列基于规则和成本的优化，如：

基于Apache Calcite 的子查询

*  项目裁剪
 *  分区裁剪
 *  过滤叠加
 *  分计划重复数据删除，以避免重复计算
 *  特殊的子查询重写，包括两部分：

1.  将其转换为左半连接
2.  将NOT IN和NOT EXISTS转换为左反连接

*  可选项 join 重排序

1.  启用`table.optimizer.join-reorder-enabled`

> **注意：**IN/EXISTS/NOT IN/NOT EXISTS目前只支持子查询重写中的连接条件。

优化器不仅基于计划，还基于数据源提供的丰富统计信息和每个算子(如io、cpu、网络和内存)的细粒度成本做出智能决策。

高级用户可以通过 CalciteConfig 对象提供自定义优化，该对象可以通过调用 TableEnvironment\#getConfig\#setPlannerConfig 提供给表环境。

## 解释表 ##

Table API 提供了一种机制来解释计算表的逻辑和优化查询计划。这是通过 TableEnvironment.explain(table)方法或 TableEnvironment.explain()方法完成的。explain(table)返回给定表的计划。Table.explain()返回多 sink 计划的结果，主要用于Blink planner。它返回一个字符串，描述三个计划：

1.  关系查询的抽象语法树，即，未经优化的逻辑查询计划
2.  优化的逻辑查询计划
3.  物理执行计划

下面的代码显示了一个示例和使用explain(Table)给出的表的相应输出：

val env = StreamExecutionEnvironment.getExecutionEnvironment
    val 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)
    
    val explanation: String = tEnv.explain(table)
    println(explanation)

== Abstract Syntax Tree ==
    LogicalUnion(all=[true])
      LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
        FlinkLogicalDataStreamScan(id=[1], fields=[count, word])
      FlinkLogicalDataStreamScan(id=[2], fields=[count, word])
    
    == Optimized Logical Plan ==
    DataStreamUnion(all=[true], union all=[count, word])
      DataStreamCalc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
        DataStreamScan(id=[1], fields=[count, word])
      DataStreamScan(id=[2], fields=[count, word])
    
    == Physical Execution Plan ==
    Stage 1 : Data Source
    	content : collect elements with CollectionInputFormat
    
    Stage 2 : Data Source
    	content : collect elements with CollectionInputFormat
    
    	Stage 3 : Operator
    		content : from: (count, word)
    		ship_strategy : REBALANCE
    
    		Stage 4 : Operator
    			content : where: (LIKE(word, _UTF-16LE'F%')), select: (count, word)
    			ship_strategy : FORWARD
    
    			Stage 5 : Operator
    				content : from: (count, word)
    				ship_strategy : REBALANCE

val settings = EnvironmentSettings.newInstance.useBlinkPlanner.inStreamingMode.build
    val tEnv = TableEnvironment.create(settings)
    
    val fieldNames = Array("count", "word")
    val fieldTypes = Array[TypeInformation[_]](Types.INT, Types.STRING)
    tEnv.registerTableSource("MySource1", new CsvTableSource("/source/path1", fieldNames, fieldTypes))
    tEnv.registerTableSource("MySource2", new CsvTableSource("/source/path2",fieldNames, fieldTypes))
    tEnv.registerTableSink("MySink1", new CsvTableSink("/sink/path1").configure(fieldNames, fieldTypes))
    tEnv.registerTableSink("MySink2", new CsvTableSink("/sink/path2").configure(fieldNames, fieldTypes))
    
    val table1 = tEnv.scan("MySource1").where("LIKE(word, 'F%')")
    table1.insertInto("MySink1")
    
    val table2 = table1.unionAll(tEnv.scan("MySource2"))
    table2.insertInto("MySink2")
    
    val explanation = tEnv.explain(false)
    println(explanation)

多 sink 计划的结果是：

== Abstract Syntax Tree ==
    LogicalSink(name=[MySink1], fields=[count, word])
    +- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
       +- LogicalTableScan(table=[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]])
    
    LogicalSink(name=[MySink2], fields=[count, word])
    +- LogicalUnion(all=[true])
       :- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
       :  +- LogicalTableScan(table=[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]])
       +- LogicalTableScan(table=[[default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]]])
    
    == Optimized Logical Plan ==
    Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')], reuse_id=[1])
    +- TableSourceScan(table=[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]], fields=[count, word])
    
    Sink(name=[MySink1], fields=[count, word])
    +- Reused(reference_id=[1])
    
    Sink(name=[MySink2], fields=[count, word])
    +- Union(all=[true], union=[count, word])
       :- Reused(reference_id=[1])
       +- TableSourceScan(table=[[default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]]], fields=[count, word])
    
    == Physical Execution Plan ==
    Stage 1 : Data Source
    	content : collect elements with CollectionInputFormat
    
    	Stage 2 : Operator
    		content : CsvTableSource(read fields: count, word)
    		ship_strategy : REBALANCE
    
    		Stage 3 : Operator
    			content : SourceConversion(table:Buffer(default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]), fields:(count, word))
    			ship_strategy : FORWARD
    
    			Stage 4 : Operator
    				content : Calc(where: (word LIKE _UTF-16LE'F%'), select: (count, word))
    				ship_strategy : FORWARD
    
    				Stage 5 : Operator
    					content : SinkConversionToRow
    					ship_strategy : FORWARD
    
    					Stage 6 : Operator
    						content : Map
    						ship_strategy : FORWARD
    
    Stage 8 : Data Source
    	content : collect elements with CollectionInputFormat
    
    	Stage 9 : Operator
    		content : CsvTableSource(read fields: count, word)
    		ship_strategy : REBALANCE
    
    		Stage 10 : Operator
    			content : SourceConversion(table:Buffer(default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]), fields:(count, word))
    			ship_strategy : FORWARD
    
    			Stage 12 : Operator
    				content : SinkConversionToRow
    				ship_strategy : FORWARD
    
    				Stage 13 : Operator
    					content : Map
    					ship_strategy : FORWARD
    
    					Stage 7 : Data Sink
    						content : Sink: CsvTableSink(count, word)
    						ship_strategy : FORWARD
    
    						Stage 14 : Data Sink
    							content : Sink: CsvTableSink(count, word)
    							ship_strategy : FORWARD

[Flink_API_DataSet_DataStream]: https://blog.csdn.net/zpf_940810653842/article/details/99636978#Java%20POJOs