Flink (十三) ：Table API 与 DataStream API 的转换 （一）

Table API 和 DataStream API 在定义数据处理管道时同样重要。DataStream API 提供了流处理的基本操作（即时间、状态和数据流管理），并且是一个相对低级的命令式编程 API。而 Table API 抽象了许多内部实现，提供了一个结构化和声明式的 API。这两个 API 都可以处理有界流和无界流。

有界流需要在处理历史数据时进行管理。无界流通常出现在实时处理场景中，可能会先通过历史数据初始化。为了高效执行，两个 API 都提供了通过优化的批处理执行模式来处理有界流。然而，由于批处理只是流处理的一个特殊情况，处理有界流的管道也可以在常规流处理模式下运行。在一个 API 中可以定义端到端的管道，而无需依赖另一个 API。然而，混合使用这两个 API 在某些情况下可能会有益：

• 在实现 DataStream API 的主要管道之前，使用 Table API 访问目录或连接外部系统。
• 在实现 DataStream API 的主要管道之前，使用一些 SQL 函数进行无状态的数据归一化和清洗。
• 如果需要更低级的操作（例如自定义定时器处理），可以切换到 DataStream API，因为 Table API 中没有提供这些功能。
• Flink 提供了特别的桥接功能，使得与 DataStream API 的集成尽可能顺畅。

1. Converting between DataStream and Table

Flink 提供了一个专门的 StreamTableEnvironment 用于与 DataStream API 集成。这个环境在常规的 TableEnvironment 基础上扩展了额外的方法，并以 DataStream API 中使用的 StreamExecutionEnvironment 作为参数。

以下代码展示了如何在两个 API 之间来回转换。表的列名和类型会自动从 DataStream 的 TypeInformation 中推导出来。由于 DataStream API 本身不支持变更日志处理，因此在流到表和表到流的转换过程中，假设使用的是仅追加/仅插入的语义。

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 org.apache.flink.types.Row;// create environments of both APIs
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);// create a DataStream
DataStream<String> dataStream = env.fromElements("Alice", "Bob", "John");// interpret the insert-only DataStream as a Table
Table inputTable = tableEnv.fromDataStream(dataStream);// register the Table object as a view and query it
tableEnv.createTemporaryView("InputTable", inputTable);
Table resultTable = tableEnv.sqlQuery("SELECT UPPER(f0) FROM InputTable");// interpret the insert-only Table as a DataStream again
DataStream<Row> resultStream = tableEnv.toDataStream(resultTable);// add a printing sink and execute in DataStream API
resultStream.print();
env.execute();// prints:
// +I[ALICE]
// +I[BOB]
// +I[JOHN]

根据查询的类型，在许多情况下，结果动态表是一个管道，在将 Table 转换为 DataStream 时，不仅会生成仅插入的变化，还可能生成撤回和其他类型的更新。在表到流的转换过程中，这可能会导致类似的异常。表接收器 'Unregistered_DataStream_Sink' 不支持消费更新变化 [...] ，在这种情况下，需要重新审查查询或切换到 toChangelogStream。

以下示例展示了如何转换更新表。每个结果行表示变更日志中的一条记录，变更标志可以通过调用 row.getKind() 来查询。在示例中，Alice 的第二个分数会在变更前创建一个更新标志 (-U)，并在变更后创建一个更新标志 (+U)。

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 org.apache.flink.types.Row;// create environments of both APIs
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);// create a DataStream
DataStream<Row> dataStream = env.fromElements(Row.of("Alice", 12),Row.of("Bob", 10),Row.of("Alice", 100));// interpret the insert-only DataStream as a Table
Table inputTable = tableEnv.fromDataStream(dataStream).as("name", "score");// register the Table object as a view and query it
// the query contains an aggregation that produces updates
tableEnv.createTemporaryView("InputTable", inputTable);
Table resultTable = tableEnv.sqlQuery("SELECT name, SUM(score) FROM InputTable GROUP BY name");// interpret the updating Table as a changelog DataStream
DataStream<Row> resultStream = tableEnv.toChangelogStream(resultTable);// add a printing sink and execute in DataStream API
resultStream.print();
env.execute();// prints:
// +I[Alice, 12]
// +I[Bob, 10]
// -U[Alice, 12]
// +U[Alice, 112]

上面的示例展示了如何通过持续地逐行发布更新来增量地计算最终结果。然而，在输入流是有限的（即有界）的情况下，可以利用批处理原则更高效地计算结果。

在批处理处理中，操作符可以在多个阶段中执行，这些阶段会先消费整个输入表，然后再发出结果。例如，连接操作符可以在执行实际连接之前对两个有界输入进行排序（即排序合并连接算法），或者在消费另一个输入之前从一个输入构建哈希表（即哈希连接算法的构建/探测阶段）。

DataStream API 和 Table API 都提供了专门的批处理运行时模式。

以下示例说明了通过简单地切换标志，统一管道能够同时处理批处理和流数据。

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;// setup DataStream API
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// set the batch runtime mode
env.setRuntimeMode(RuntimeExecutionMode.BATCH);// uncomment this for streaming mode
// env.setRuntimeMode(RuntimeExecutionMode.STREAMING);// setup Table API
// the table environment adopts the runtime mode during initialization
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);// define the same pipeline as above// prints in BATCH mode:
// +I[Bob, 10]
// +I[Alice, 112]// prints in STREAMING mode:
// +I[Alice, 12]
// +I[Bob, 10]
// -U[Alice, 12]
// +U[Alice, 112]

一旦变更日志应用到外部系统（例如键值存储），可以看到两种模式都能够产生完全相同的输出表。通过在发出结果之前消费所有输入数据，批处理模式的变更日志仅包含插入类型的变化。

1.1 Dependencies and Imports

结合 Table API 和 DataStream API 的项目需要添加以下其中一个桥接模块。这些模块包括了 flink-table-api-java 或 flink-table-api-scala 的传递依赖，以及相应语言特定的 DataStream API 模块。

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java-bridge_2.12</artifactId><version>1.20.0</version><scope>provided</scope>
</dependency>

以下导入是声明使用 Java 或 Scala 版本的 DataStream API 和 Table API 的公共管道所必需的。

// imports for Java DataStream API
import org.apache.flink.streaming.api.*;
import org.apache.flink.streaming.api.environment.*;// imports for Table API with bridging to Java DataStream API
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.*;

1.2 Configuration

TableEnvironment 将采用传入的 StreamExecutionEnvironment 的所有配置选项。然而，在实例化之后，无法保证对 StreamExecutionEnvironment 配置的进一步更改会传播到 StreamTableEnvironment。建议在切换到 Table API 之前，尽早在 DataStream API 中设置所有配置选项。

import java.time.ZoneId;
import org.apache.flink.core.execution.CheckpointingMode.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;// create Java DataStream APIStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// set various configuration earlyenv.setMaxParallelism(256);env.getConfig().addDefaultKryoSerializer(MyCustomType.class, CustomKryoSerializer.class);env.getCheckpointConfig().setCheckpointingConsistencyMode(CheckpointingMode.EXACTLY_ONCE);// then switch to Java Table APIStreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);// set configuration earlytableEnv.getConfig().setLocalTimeZone(ZoneId.of("Europe/Berlin"));// start defining your pipelines in both APIs...

1.3 Execution Behavior 

两个 API 都提供了执行管道的方法。换句话说：如果需要，它们会编译一个作业图并提交到集群，触发执行。结果将被流式传输到声明的接收器。通常，两个 API 都在方法名称中使用 execute 来标识这种行为。然而，Table API 和 DataStream API 之间的执行行为略有不同。

1.3.1 DataStream API

DataStream API 的 StreamExecutionEnvironment 使用构建器来构建复杂的管道。该管道可能会拆分成多个分支，这些分支可能会或可能不会以Sink结束。执行环境会缓冲所有这些定义的分支，直到提交作业为止。StreamExecutionEnvironment.execute() 提交整个构建的管道，并在之后清除构建器。换句话说：不再声明Source和Sink，可以将新的管道添加到构建器中。因此，每个 DataStream 程序通常以调用 StreamExecutionEnvironment.execute() 结束。或者，DataStream.executeAndCollect() 会隐式地为结果流式传输到本地客户端定义一个Sink

1.3.2 Table API

在 Table API 中，分支管道仅在 StatementSet 中受到支持，其中每个分支必须声明一个最终的Sink。TableEnvironment 和 StreamTableEnvironment 都不提供专门的通用 execute() 方法。相反，它们提供了提交单个Source到Sink管道或StatementSet的方法：

// execute with explicit sink
tableEnv.from("InputTable").insertInto("OutputTable").execute();tableEnv.executeSql("INSERT INTO OutputTable SELECT * FROM InputTable");tableEnv.createStatementSet().add(tableEnv.from("InputTable").insertInto("OutputTable")).add(tableEnv.from("InputTable").insertInto("OutputTable2")).execute();tableEnv.createStatementSet().addInsertSql("INSERT INTO OutputTable SELECT * FROM InputTable").addInsertSql("INSERT INTO OutputTable2 SELECT * FROM InputTable").execute();// execute with implicit local sinktableEnv.from("InputTable").execute().print();tableEnv.executeSql("SELECT * FROM InputTable").print();

为了结合这两种执行行为，每次调用 StreamTableEnvironment.toDataStream 或 StreamTableEnvironment.toChangelogStream 时，会将 Table API 子管道物化（即编译）并插入到 DataStream API 管道构建器中。这意味着之后必须调用 StreamExecutionEnvironment.execute() 或 DataStream.executeAndCollect。在 Table API 中的执行不会触发这些“外部部分”。

// (1)// adds a branch with a printing sink to the StreamExecutionEnvironment
tableEnv.toDataStream(table).print();// (2)// executes a Table API end-to-end pipeline as a Flink job and prints locally,
// thus (1) has still not been executed
table.execute().print();// executes the DataStream API pipeline with the sink defined in (1) as a
// Flink job, (2) was already running before
env.execute();

2. Batch Runtime Mode

批处理运行时模式是针对有界 Flink 程序的专门执行模式。一般来说，有界性是数据源的一个属性，它告诉我们该数据源中所有记录在执行前是否已知，或者是否会出现新的数据，可能是无限期的。一个作业，如果其所有数据源都是有界的，则该作业是有界的，否则就是无界的。另一方面，流处理运行时模式可以用于有界和无界作业。Table API 和 SQL planner为这两种模式提供了一组专门的优化器规则和运行时操作符。目前，运行时模式不会自动从数据源推导出来，因此，必须显式设置，或者在实例化 StreamTableEnvironment 时从 StreamExecutionEnvironment 继承。

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;// adopt mode from StreamExecutionEnvironment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);// or// set mode explicitly for StreamTableEnvironment
// it will be propagated to StreamExecutionEnvironment during planning
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, EnvironmentSettings.inBatchMode());

在将运行时模式设置为 BATCH 之前，必须满足以下前提条件：

1. 所有数据源必须声明自己为有界的。
2. 当前，表数据源必须仅发出插入（insert-only）变化。
3. 操作符需要足够的堆外内存来进行排序和处理其他中间结果。
4. 所有表操作必须在批处理模式下可用。当前，某些操作仅在流处理模式下可用。

批处理执行具有以下含义（其中包括但不限于）： 

• 操作符中不会生成或使用渐进watermarks。然而，数据源在关闭之前会发出一个最大watermarks。
• 任务之间的交换可能是阻塞的，这取决于 execution.batch-shuffle-mode。这也意味着与在流处理模式下执行相同管道相比，可能会需要更少的资源。
• 检查点（Checkpointing）被禁用。插入了人工状态后端（Artificial state backends）。
• 表操作不会产生增量更新，而是只生成一个完整的最终结果，并转换为一个仅插入的变更日志流。

由于批处理可以视为流处理的特例，我们建议首先实现一个流处理管道，因为它是有界和无界数据的最通用实现。理论上，流处理管道可以执行所有操作符。然而，实际上，某些操作可能没有太大意义，因为它们会导致状态不断增长，因此不被支持。全局排序就是一个例子，它仅在批处理模式下可用。简而言之：应该可以在批处理模式下运行一个正常工作的流处理管道，但不一定能够反过来运行。

以下示例展示了如何使用 DataGen 表数据源在批处理模式下进行操作。许多数据源提供的选项可以隐式地使连接器变为有界，例如通过定义终止偏移量或时间戳。在我们的示例中，我们通过 number-of-rows 选项限制了行数。

import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;Table table =tableEnv.from(TableDescriptor.forConnector("datagen").option("number-of-rows", "10") // make the source bounded.schema(Schema.newBuilder().column("uid", DataTypes.TINYINT()).column("payload", DataTypes.STRING()).build()).build());// convert the Table to a DataStream and further transform the pipeline
tableEnv.toDataStream(table).keyBy(r -> r.<Byte>getFieldAs("uid")).map(r -> "My custom operator: " + r.<String>getFieldAs("payload")).executeAndCollect().forEachRemaining(System.out::println);// prints:
// My custom operator: 9660912d30a43c7b035e15bd...
// My custom operator: 29f5f706d2144f4a4f9f52a0...
// ...

2.1 Changelog Unification

在大多数情况下，从流处理模式切换到批处理模式，反之亦然，管道定义本身在 Table API 和 DataStream API 中可以保持不变。然而，如前所述，由于批处理模式避免了增量操作，最终的变更日志流可能会有所不同。依赖于事件时间并利用水印作为完整性标记的基于时间的操作，能够生成一个仅插入的变更日志流，这与运行时模式无关。

以下 Java 示例展示了一个 Flink 程序，它不仅在 API 层面上是统一的，而且在最终生成的变更日志流中也是统一的。该示例使用基于两张表（UserTable 和 OrderTable）中时间属性（ts）的区间连接（interval join）来连接两张表。它使用 DataStream API 实现了一个自定义操作符，该操作符通过 KeyedProcessFunction 和值状态去重用户名。

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import java.time.LocalDateTime;// setup DataStream API
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// use BATCH or STREAMING mode
env.setRuntimeMode(RuntimeExecutionMode.BATCH);// setup Table API
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);// create a user stream
DataStream<Row> userStream = env.fromElements(Row.of(LocalDateTime.parse("2021-08-21T13:00:00"), 1, "Alice"),Row.of(LocalDateTime.parse("2021-08-21T13:05:00"), 2, "Bob"),Row.of(LocalDateTime.parse("2021-08-21T13:10:00"), 2, "Bob")).returns(Types.ROW_NAMED(new String[] {"ts", "uid", "name"},Types.LOCAL_DATE_TIME, Types.INT, Types.STRING));// create an order stream
DataStream<Row> orderStream = env.fromElements(Row.of(LocalDateTime.parse("2021-08-21T13:02:00"), 1, 122),Row.of(LocalDateTime.parse("2021-08-21T13:07:00"), 2, 239),Row.of(LocalDateTime.parse("2021-08-21T13:11:00"), 2, 999)).returns(Types.ROW_NAMED(new String[] {"ts", "uid", "amount"},Types.LOCAL_DATE_TIME, Types.INT, Types.INT));// create corresponding tables
tableEnv.createTemporaryView("UserTable",userStream,Schema.newBuilder().column("ts", DataTypes.TIMESTAMP(3)).column("uid", DataTypes.INT()).column("name", DataTypes.STRING()).watermark("ts", "ts - INTERVAL '1' SECOND").build());tableEnv.createTemporaryView("OrderTable",orderStream,Schema.newBuilder().column("ts", DataTypes.TIMESTAMP(3)).column("uid", DataTypes.INT()).column("amount", DataTypes.INT()).watermark("ts", "ts - INTERVAL '1' SECOND").build());// perform interval join
Table joinedTable =tableEnv.sqlQuery("SELECT U.name, O.amount " +"FROM UserTable U, OrderTable O " +"WHERE U.uid = O.uid AND O.ts BETWEEN U.ts AND U.ts + INTERVAL '5' MINUTES");DataStream<Row> joinedStream = tableEnv.toDataStream(joinedTable);joinedStream.print();// implement a custom operator using ProcessFunction and value state
joinedStream.keyBy(r -> r.<String>getFieldAs("name")).process(new KeyedProcessFunction<String, Row, String>() {ValueState<String> seen;@Overridepublic void open(OpenContext openContext) {seen = getRuntimeContext().getState(new ValueStateDescriptor<>("seen", String.class));}@Overridepublic void processElement(Row row, Context ctx, Collector<String> out)throws Exception {String name = row.getFieldAs("name");if (seen.value() == null) {seen.update(name);out.collect(name);}}}).print();// execute unified pipeline
env.execute();// prints (in both BATCH and STREAMING mode):
// +I[Bob, 239]
// +I[Alice, 122]
// +I[Bob, 999]
//
// Bob
// Alice