[大数据]（十八）Flink Table API & SQL 编程指南 Table API 和Datastream API 集成

在定义数据处理管道时，Table API 和 DataStream API 同样重要。

DataStream API 在一个相对较低级别的命令式编程 API 中提供了流处理的原语（即时间、状态和数据流管理）。Table API 抽象了许多内部结构，并提供了结构化和声明性的 API。

两种 API 都可以处理有界和无界流。

处理历史数据时需要管理有界流。无限流发生在可能首先用历史数据初始化的实时处理场景中。

为了高效执行，两个 API 都以优化的批处理执行模式提供处理有界流。但是，由于批处理只是流的一种特殊情况，因此也可以在常规流执行模式下运行有界流的管道。

一个 API 中的管道可以端到端定义，而不依赖于另一个 API。但是，出于各种原因，混合使用这两种 API 可能会很有用：

• 在 DataStream API 中实现主管道之前，使用表生态系统轻松访问目录或连接到外部系统。
• 在 DataStream API 中实现主管道之前，访问一些用于无状态数据规范化和清理的 SQL 函数。
• 如果 Table API 中不存在更底层的操作（例如自定义计时器处理），请不要切换到 DataStream API。

Flink 提供了特殊的桥接功能，使与 DataStream API 的集成尽可能顺畅。

在 DataStream 和 Table API 之间切换会增加一些转换开销。例如，RowData部分处理二进制数据的表运行时的内部数据结构（即）需要转换为对用户更友好的数据结构（即Row）。通常，可以忽略此开销，但为了完整起见，此处提及。

DataStream 和 Table 之间的转换

Flink 提供了一个专门StreamTableEnvironment用于与 DataStream 集成的 API。TableEnvironment这些环境使用其他方法扩展常规，StreamExecutionEnvironment并将 DataStream API 中使用的作为参数。

以下代码显示了如何在两个 API 之间来回切换的示例。的列名和类型Table自动派生自TypeInformation的DataStream。由于 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]

fromDataStream和的完整语义toDataStream可以在下面的专门部分中找到。特别是，本节讨论了如何使用更复杂和嵌套的类型来影响模式派生。它还涵盖了使用事件时间和水印。

根据查询的类型，在许多情况下，生成的动态表是一个管道，它不仅在将转换Table为 一个DataStream时产生仅插入更改，DataStream而且还会产生撤回和其他类型的更新。在表到流的转换过程中，这可能会导致类似于以下的异常

Table sink 'Unregistered_DataStream_Sink_1' doesn't support consuming update changes [...].

在这种情况下，需要再次修改查询或切换到toChangelogStream.

以下示例显示了如何转换更新表。每个结果行都代表更改日志中的一个条目，该条目带有一个可以通过调用row.getKind()它来查询的更改标志。在示例中，第二个分数在(U )之前Alice创建更新，在( 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]

fromChangelogStream和的完整语义toChangelogStream可以在下面的专门部分中找到。特别是，本节讨论了如何使用更复杂和嵌套的类型来影响模式派生。它涵盖了使用事件时间和水印。它讨论了如何为输入和输出流声明主键和更改日志模式。

上面的示例显示了如何通过为每个传入记录连续发出逐行更新来增量计算最终结果。然而，在输入流是有限的（即有界的）的情况下，可以通过利用批处理原理更有效地计算结果。

在批处理中，运算符可以在连续阶段执行，在发出结果之前消耗整个输入表。例如，连接运算符可以在执行实际连接之前对两个有界输入进行排序（即排序合并连接算法），或者在使用另一个输入之前从一个输入构建哈希表（即哈希连接算法的构建/探测阶段）。

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]

一旦将变更日志应用到外部系统（例如键值存储），可以看到两种模式都能够生成完全相同的输出表。通过在发出结果之前消耗所有输入数据，批处理模式的更改日志仅包含仅插入更改。另请参阅下面的专用批处理模式部分以获取更多信息。

依赖项和导入

将 Table API 与 DataStream API 结合的项目需要添加以下桥接模块之一。flink-table-api-java它们包括对orflink-table-api-scala和相应语言特定的 DataStream API 模块的传递依赖。

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

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

// 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.*;

配置

将TableEnvironment采用传递的所有配置选项StreamExecutionEnvironment。但是，不能保证对配置的进一步更改StreamExecutionEnvironment 会传播到StreamTableEnvironment其实例化之后。从 Table API 到 DataStream API 的选项传播发生在规划期间。

我们建议在切换到 Table API 之前尽早在 DataStream API 中设置所有配置选项。

import java.time.ZoneId;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// create Java DataStream API

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// set various configuration early

env.setMaxParallelism(256);

env.getConfig().addDefaultKryoSerializer(MyCustomType.class, CustomKryoSerializer.class);

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// then switch to Java Table API

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// set configuration early

tableEnv.getConfig().setLocalTimeZone(ZoneId.of("Europe/Berlin"));

// start defining your pipelines in both APIs...

执行行为

这两个 API 都提供了执行管道的方法。换句话说：如果需要，他们会编译一个作业图，该作业图将提交到集群并触发执行。结果将流式传输到声明的接收器。

execute通常，两个 API 都使用方法名称中的术语来标记此类行为。但是，Table API 和 DataStream API 的执行行为略有不同。

datastream API

DataStream API StreamExecutionEnvironment使用构建器模式来构建复杂的管道。管道可能会分成多个分支，这些分支可能会或可能不会以接收器结束。环境缓冲所有这些定义的分支，直到提交作业。

StreamExecutionEnvironment.execute()提交整个构建的管道并随后清除构建器。换句话说：不再声明源和接收器，并且可以将新管道添加到构建器中。因此，每个 DataStream 程序通常都以调用StreamExecutionEnvironment.execute(). 或者，DataStream.executeAndCollect()隐式定义一个接收器，用于将结果流式传输到本地客户端。

table API

StatementSet在 Table API 中，仅在每个分支必须声明最终接收器的情况下才支持分支管道。两者TableEnvironment都StreamTableEnvironment没有提供专用的通用execute()方法。相反，它们提供了提交单个源到接收器管道或语句集的方法：

// 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 sink

tableEnv.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();

批处理运行时模式

批处理运行时模式是有界Flink 程序的专用执行模式。

一般来说，有界性是数据源的一个属性，它告诉我们来自该源的所有记录在执行之前是否已知，或者是否会无限期地显示新数据。反过来，如果一个工作的所有来源都是有界的，那么它是有界的，否则是无界的。

另一方面，流式运行时模式可用于有界和无界作业。

有关不同执行模式的更多信息，另请参阅相应的DataStream API 部分。

Table API & SQL 规划器为这两种模式中的任何一种都提供了一组专门的优化器规则和运行时运算符。

目前，运行时模式不是从源自动派生的，因此，它必须显式设置或StreamExecutionEnvironment在实例化 采用StreamTableEnvironment：

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 之前，必须满足以下先决条件：

• 所有来源都必须声明自己是有界的。

• 目前，表源必须发出仅插入更改。

• 运算符需要足够量的堆外内存 来进行排序和其他中间结果。

• 所有表操作都必须在批处理模式下可用。目前，其中一些仅在流模式下可用。请查看相应的 Table API & SQL 页面。

批处理执行具有以下含义（除其他外）：

• 渐进式水印既不会生成也不会在运算符中使用。但是，源在关闭之前会发出最大水印。

• 根据execution.batch-shuffle-mode. 与在流模式下执行相同的管道相比，这也意味着可能需要更少的资源。

• 检查点被禁用。插入了人工状态后端。

• 表操作不会产生增量更新，而只会产生一个完整的最终结果，该结果会转换为仅插入的变更日志流。

由于批处理可以被视为流处理的一种特殊情况，因此我们建议首先实现流式管道，因为它是有界和无界数据的最通用实现。

理论上，流式管道可以执行所有运算符。但是，在实践中，某些操作可能没有多大意义，因为它们会导致状态不断增长，因此不受支持。全局排序是一个仅在批处理模式下可用的示例。简而言之：应该可以在批处理模式下运行工作流管道，但反之亦然。

下面的示例展示了如何使用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...
// ...

Changelog统一

在大多数情况下，当从流模式切换到批处理模式时，管道定义本身可以在 Table API 和 DataStream API 中保持不变，反之亦然。但是，如前所述，由于避免了批处理模式下的增量操作，生成的变更日志流可能会有所不同。

依赖于事件时间并利用水印作为完整性标记的基于时间的操作能够生成独立于运行时模式的仅插入更改日志流。

下面的 Java 示例说明了一个 Flink 程序，该程序不仅在 API 级别上统一，而且在生成的更改日志流中也统一。该示例使用基于两个表中的时间属性 （ts） 的间隔连接来联接 SQL 中的两个表（UserTable 和 OrderTable）。它使用 DataStream API 实现一个自定义运算符，该运算符使用KeyedProcessFunction and value state对用户名进行重复数据删除。

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.configuration.Configuration;
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;

// 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;

          @Override
          public void open(Configuration parameters) {
              seen = getRuntimeContext().getState(
                  new ValueStateDescriptor<>("seen", String.class));
          }

          @Override
          public 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

处理（仅插入）流 (Handling of (Insert-Only) Streams)

StreamTableEnvironment提供以下方法来转换 DataStream API：

• fromDataStream(DataStream)：将仅插入更改和任意类型的流解释为表。默认情况下不传播事件时间和水印。

• fromDataStream(DataStream, Schema)：将仅插入更改和任意类型的流解释为表。可选模式允许丰富列数据类型并添加时间属性、水印策略、其他计算列或主键。

• createTemporaryView(String, DataStream)：以名称注册流以在 SQL 中访问它。它是createTemporaryView(String, fromDataStream(DataStream)).

• createTemporaryView(String, DataStream, Schema)：以名称注册流以在 SQL 中访问它。它是createTemporaryView(String, fromDataStream(DataStream, Schema)).

• toDataStream(Table)：将表转换为仅插入更改的流。默认流记录类型是org.apache.flink.types.Row. 单个行时间属性列被写回到 DataStream API 的记录中。水印也被传播。

• toDataStream(Table, AbstractDataType)：将表转换为仅插入更改的流。此方法接受一种数据类型来表达所需的流记录类型。规划器可能会插入隐式强制转换和重新排序列以将列映射到（可能是嵌套的）数据类型的字段。

• toDataStream(Table, Class)：toDataStream(Table, DataTypes.of(Class)) 快速创建所需数据类型的快捷方式。

从 Table API 的角度来看，与 DataStream API 之间的转换类似于读取或写入已使用 SQL中的CREATE TABLEDDL定义的虚拟表连接器。

虚拟CREATE TABLE name (schema) WITH (options)语句中的模式部分可以从 DataStream 的类型信息中自动派生、丰富或完全使用 org.apache.flink.table.api.Schema.

虚拟 DataStream 表连接器为每一行公开以下元数据：

虚拟 DataStream 表源实现SupportsSourceWatermark 并因此允许调用SOURCE_WATERMARK()内置函数作为水印策略以采用来自 DataStream API 的水印。

fromDataStream

下面的代码展示了如何fromDataStream用于不同的场景。

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import java.time.Instant;

// some example POJO
public static class User {
  public String name;

  public Integer score;

  public Instant event_time;

  // default constructor for DataStream API
  public User() {}

  // fully assigning constructor for Table API
  public User(String name, Integer score, Instant event_time) {
    this.name = name;
    this.score = score;
    this.event_time = event_time;
  }
}

// create a DataStream
DataStream<User> dataStream =
    env.fromElements(
        new User("Alice", 4, Instant.ofEpochMilli(1000)),
        new User("Bob", 6, Instant.ofEpochMilli(1001)),
        new User("Alice", 10, Instant.ofEpochMilli(1002)));


// === EXAMPLE 1 ===

// derive all physical columns automatically

Table table = tableEnv.fromDataStream(dataStream);
table.printSchema();
// prints:
// (
//  `name` STRING,
//  `score` INT,
//  `event_time` TIMESTAMP_LTZ(9)
// )


// === EXAMPLE 2 ===

// derive all physical columns automatically
// but add computed columns (in this case for creating a proctime attribute column)

Table table = tableEnv.fromDataStream(
    dataStream,
    Schema.newBuilder()
        .columnByExpression("proc_time", "PROCTIME()")
        .build());
table.printSchema();
// prints:
// (
//  `name` STRING,
//  `score` INT NOT NULL,
//  `event_time` TIMESTAMP_LTZ(9),
//  `proc_time` TIMESTAMP_LTZ(3) NOT NULL *PROCTIME* AS PROCTIME()
//)


// === EXAMPLE 3 ===

// derive all physical columns automatically
// but add computed columns (in this case for creating a rowtime attribute column)
// and a custom watermark strategy

Table table =
    tableEnv.fromDataStream(
        dataStream,
        Schema.newBuilder()
            .columnByExpression("rowtime", "CAST(event_time AS TIMESTAMP_LTZ(3))")
            .watermark("rowtime", "rowtime - INTERVAL '10' SECOND")
            .build());
table.printSchema();
// prints:
// (
//  `name` STRING,
//  `score` INT,
//  `event_time` TIMESTAMP_LTZ(9),
//  `rowtime` TIMESTAMP_LTZ(3) *ROWTIME* AS CAST(event_time AS TIMESTAMP_LTZ(3)),
//  WATERMARK FOR `rowtime`: TIMESTAMP_LTZ(3) AS rowtime - INTERVAL '10' SECOND
// )


// === EXAMPLE 4 ===

// derive all physical columns automatically
// but access the stream record's timestamp for creating a rowtime attribute column
// also rely on the watermarks generated in the DataStream API

// we assume that a watermark strategy has been defined for `dataStream` before
// (not part of this example)
Table table =
    tableEnv.fromDataStream(
        dataStream,
        Schema.newBuilder()
            .columnByMetadata("rowtime", "TIMESTAMP_LTZ(3)")
            .watermark("rowtime", "SOURCE_WATERMARK()")
            .build());
table.printSchema();
// prints:
// (
//  `name` STRING,
//  `score` INT,
//  `event_time` TIMESTAMP_LTZ(9),
//  `rowtime` TIMESTAMP_LTZ(3) *ROWTIME* METADATA,
//  WATERMARK FOR `rowtime`: TIMESTAMP_LTZ(3) AS SOURCE_WATERMARK()
// )


// === EXAMPLE 5 ===

// define physical columns manually
// in this example,
//   - we can reduce the default precision of timestamps from 9 to 3
//   - we also project the columns and put `event_time` to the beginning

Table table =
    tableEnv.fromDataStream(
        dataStream,
        Schema.newBuilder()
            .column("event_time", "TIMESTAMP_LTZ(3)")
            .column("name", "STRING")
            .column("score", "INT")
            .watermark("event_time", "SOURCE_WATERMARK()")
            .build());
table.printSchema();
// prints:
// (
//  `event_time` TIMESTAMP_LTZ(3) *ROWTIME*,
//  `name` VARCHAR(200),
//  `score` INT
// )
// note: the watermark strategy is not shown due to the inserted column reordering projection

示例 1 说明了一个不需要基于时间的操作的简单用例。

示例 4 是最常见的用例，当基于时间的操作（例如窗口或间隔连接）应成为管道的一部分时。示例 2 是这些基于时间的操作应该在处理时间内工作的最常见用例。

示例 5 完全依赖于用户的声明。这对于用适当的数据类型替换来自 DataStream API（将RAW在 Table API 中）的泛型类型很有用。

由于DataType比 更丰富TypeInformation，我们可以轻松启用不可变 POJO 和其他复杂的数据结构。以下 Java 示例显示了可能的情况。另请查看 DataStream API 的 Data Types & Serialization页面以获取有关那里支持的类型的更多信息。

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;

// the DataStream API does not support immutable POJOs yet,
// the class will result in a generic type that is a RAW type in Table API by default
public static class User {

    public final String name;

    public final Integer score;

    public User(String name, Integer score) {
        this.name = name;
        this.score = score;
    }
}

// create a DataStream
DataStream<User> dataStream = env.fromElements(
    new User("Alice", 4),
    new User("Bob", 6),
    new User("Alice", 10));

// since fields of a RAW type cannot be accessed, every stream record is treated as an atomic type
// leading to a table with a single column `f0`

Table table = tableEnv.fromDataStream(dataStream);
table.printSchema();
// prints:
// (
//  `f0` RAW('User', '...')
// )

// instead, declare a more useful data type for columns using the Table API's type system
// in a custom schema and rename the columns in a following `as` projection

Table table = tableEnv
    .fromDataStream(
        dataStream,
        Schema.newBuilder()
            .column("f0", DataTypes.of(User.class))
            .build())
    .as("user");
table.printSchema();
// prints:
// (
//  `user` *User<`name` STRING,`score` INT>*
// )

// data types can be extracted reflectively as above or explicitly defined

Table table3 = tableEnv
    .fromDataStream(
        dataStream,
        Schema.newBuilder()
            .column(
                "f0",
                DataTypes.STRUCTURED(
                    User.class,
                    DataTypes.FIELD("name", DataTypes.STRING()),
                    DataTypes.FIELD("score", DataTypes.INT())))
            .build())
    .as("user");
table.printSchema();
// prints:
// (
//  `user` *User<`name` STRING,`score` INT>*
// )

createTemporaryView

DataStream可以直接注册为视图（可能通过模式丰富）。

创建的视图DataStream只能注册为临时视图。由于它们的内联/匿名 性质，无法将它们注册到永久catalog中。
下面的代码展示了如何createTemporaryView用于不同的场景。

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;

// create some DataStream
DataStream<Tuple2<Long, String>> dataStream = env.fromElements(
    Tuple2.of(12L, "Alice"),
    Tuple2.of(0L, "Bob"));


// === EXAMPLE 1 ===

// register the DataStream as view "MyView" in the current session
// all columns are derived automatically

tableEnv.createTemporaryView("MyView", dataStream);

tableEnv.from("MyView").printSchema();

// prints:
// (
//  `f0` BIGINT NOT NULL,
//  `f1` STRING
// )


// === EXAMPLE 2 ===

// register the DataStream as view "MyView" in the current session,
// provide a schema to adjust the columns similar to `fromDataStream`

// in this example, the derived NOT NULL information has been removed

tableEnv.createTemporaryView(
    "MyView",
    dataStream,
    Schema.newBuilder()
        .column("f0", "BIGINT")
        .column("f1", "STRING")
        .build());

tableEnv.from("MyView").printSchema();

// prints:
// (
//  `f0` BIGINT,
//  `f1` STRING
// )


// === EXAMPLE 3 ===

// use the Table API before creating the view if it is only about renaming columns

tableEnv.createTemporaryView(
    "MyView",
    tableEnv.fromDataStream(dataStream).as("id", "name"));

tableEnv.from("MyView").printSchema();

// prints:
// (
//  `id` BIGINT NOT NULL,
//  `name` STRING
// )

toDataStream

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.types.Row;
import java.time.Instant;

// POJO with mutable fields
// since no fully assigning constructor is defined, the field order
// is alphabetical [event_time, name, score]
public static class User {

    public String name;

    public Integer score;

    public Instant event_time;
}

tableEnv.executeSql(
    "CREATE TABLE GeneratedTable "
    + "("
    + "  name STRING,"
    + "  score INT,"
    + "  event_time TIMESTAMP_LTZ(3),"
    + "  WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND"
    + ")"
    + "WITH ('connector'='datagen')");

Table table = tableEnv.from("GeneratedTable");


// === EXAMPLE 1 ===

// use the default conversion to instances of Row

// since `event_time` is a single rowtime attribute, it is inserted into the DataStream
// metadata and watermarks are propagated

DataStream<Row> dataStream = tableEnv.toDataStream(table);


// === EXAMPLE 2 ===

// a data type is extracted from class `User`,
// the planner reorders fields and inserts implicit casts where possible to convert internal
// data structures to the desired structured type

// since `event_time` is a single rowtime attribute, it is inserted into the DataStream
// metadata and watermarks are propagated

DataStream<User> dataStream = tableEnv.toDataStream(table, User.class);

// data types can be extracted reflectively as above or explicitly defined

DataStream<User> dataStream =
    tableEnv.toDataStream(
        table,
        DataTypes.STRUCTURED(
            User.class,
            DataTypes.FIELD("name", DataTypes.STRING()),
            DataTypes.FIELD("score", DataTypes.INT()),
            DataTypes.FIELD("event_time", DataTypes.TIMESTAMP_LTZ(3))));

请注意，仅支持非更新表toDataStream。通常，基于时间的操作（例如窗口、间隔连接或MATCH_RECOGNIZE子句）非常适合仅插入管道，以及投影和过滤器等简单操作。

具有产生更新操作的管道可以使用toChangelogStream.

处理变更日志流

在内部，Flink 的表运行时是一个变更日志处理器。概念页面描述了 动态表和流如何 相互关联。

StreamTableEnvironment提供以下方法来公开这些变更数据捕获(CDC) 功能：

• fromChangelogStream(DataStream)：将变更日志条目流解释为表格。流记录类型必须是org.apache.flink.types.Row，因为它的RowKind标志是在运行时评估的。默认情况下不传播事件时间和水印。此方法需要一个包含各种更改（在 中枚举org.apache.flink.types.RowKind）作为默认值的更改日志ChangelogMode。

• fromChangelogStream(DataStream, Schema): 允许为DataStream类似于fromDataStream(DataStream, Schema). 否则语义等于fromChangelogStream(DataStream)。

• fromChangelogStream(DataStream, Schema, ChangelogMode)：完全控制如何将流解释为变更日志。传递ChangelogMode帮助计划者区分insert-only、 upsert或retract行为。

• toChangelogStream(Table): 的反向操作fromChangelogStream(DataStream)。它在运行时生成带有实例的流org.apache.flink.types.Row并为每条记录设置RowKind标志。该方法支持各种更新表。如果输入表包含单个行时间列，它将被传播到流记录的时间戳中。水印也将被传播。

• toChangelogStream(Table, Schema): 的反向操作fromChangelogStream(DataStream, Schema)。该方法可以丰富产生的列数据类型。如有必要，计划者可能会插入隐式强制转换。可以将行时间写为元数据列。

• toChangelogStream(Table, Schema, ChangelogMode)：完全控制如何将表转换为变更日志流。传递ChangelogMode帮助计划者区分insert-only、 upsert或retract行为。

从 Table API 的角度来看，与 DataStream API 之间的转换类似于读取或写入已使用 SQL中的CREATE TABLEDDL定义的虚拟表连接器。

因为fromChangelogStream行为类似于fromDataStream，我们建议在继续之前阅读上一节。

此虚拟连接器还支持读取和写入rowtime流记录的元数据。

虚拟表源实现SupportsSourceWatermark.

fromChangelogStream

下面的代码展示了如何fromChangelogStream用于不同的场景。

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

// === EXAMPLE 1 ===

// interpret the stream as a retract stream

// create a changelog DataStream
DataStream<Row> dataStream =
    env.fromElements(
        Row.ofKind(RowKind.INSERT, "Alice", 12),
        Row.ofKind(RowKind.INSERT, "Bob", 5),
        Row.ofKind(RowKind.UPDATE_BEFORE, "Alice", 12),
        Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100));

// interpret the DataStream as a Table
Table table = tableEnv.fromChangelogStream(dataStream);

// register the table under a name and perform an aggregation
tableEnv.createTemporaryView("InputTable", table);
tableEnv
    .executeSql("SELECT f0 AS name, SUM(f1) AS score FROM InputTable GROUP BY f0")
    .print();

// prints:
// +----+--------------------------------+-------------+
// | op |                           name |       score |
// +----+--------------------------------+-------------+
// | +I |                            Bob |           5 |
// | +I |                          Alice |          12 |
// | -D |                          Alice |          12 |
// | +I |                          Alice |         100 |
// +----+--------------------------------+-------------+


// === EXAMPLE 2 ===

// interpret the stream as an upsert stream (without a need for UPDATE_BEFORE)

// create a changelog DataStream
DataStream<Row> dataStream =
    env.fromElements(
        Row.ofKind(RowKind.INSERT, "Alice", 12),
        Row.ofKind(RowKind.INSERT, "Bob", 5),
        Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100));

// interpret the DataStream as a Table
Table table =
    tableEnv.fromChangelogStream(
        dataStream,
        Schema.newBuilder().primaryKey("f0").build(),
        ChangelogMode.upsert());

// register the table under a name and perform an aggregation
tableEnv.createTemporaryView("InputTable", table);
tableEnv
    .executeSql("SELECT f0 AS name, SUM(f1) AS score FROM InputTable GROUP BY f0")
    .print();

// prints:
// +----+--------------------------------+-------------+
// | op |                           name |       score |
// +----+--------------------------------+-------------+
// | +I |                            Bob |           5 |
// | +I |                          Alice |          12 |
// | -U |                          Alice |          12 |
// | +U |                          Alice |         100 |
// +----+--------------------------------+-------------+

示例 1 中显示的默认值ChangelogMode对于大多数用例来说应该足够了，因为它接受各种更改。

但是，示例 2 显示了如何通过使用 upsert 模式将更新消息的数量减少 50% 来限制传入更改的种类以提高效率。可以通过为toChangelogStream.

toChangelogStream

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.ProcessFunction;
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.data.StringData;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import static org.apache.flink.table.api.Expressions.*;

// create Table with event-time
tableEnv.executeSql(
    "CREATE TABLE GeneratedTable "
    + "("
    + "  name STRING,"
    + "  score INT,"
    + "  event_time TIMESTAMP_LTZ(3),"
    + "  WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND"
    + ")"
    + "WITH ('connector'='datagen')");

Table table = tableEnv.from("GeneratedTable");


// === EXAMPLE 1 ===

// convert to DataStream in the simplest and most general way possible (no event-time)

Table simpleTable = tableEnv
    .fromValues(row("Alice", 12), row("Alice", 2), row("Bob", 12))
    .as("name", "score")
    .groupBy($("name"))
    .select($("name"), $("score").sum());

tableEnv
    .toChangelogStream(simpleTable)
    .executeAndCollect()
    .forEachRemaining(System.out::println);

// prints:
// +I[Bob, 12]
// +I[Alice, 12]
// -U[Alice, 12]
// +U[Alice, 14]


// === EXAMPLE 2 ===

// convert to DataStream in the simplest and most general way possible (with event-time)

DataStream<Row> dataStream = tableEnv.toChangelogStream(table);

// since `event_time` is a single time attribute in the schema, it is set as the
// stream record's timestamp by default; however, at the same time, it remains part of the Row

dataStream.process(
    new ProcessFunction<Row, Void>() {
        @Override
        public void processElement(Row row, Context ctx, Collector<Void> out) {

             // prints: [name, score, event_time]
             System.out.println(row.getFieldNames(true));

             // timestamp exists twice
             assert ctx.timestamp() == row.<Instant>getFieldAs("event_time").toEpochMilli();
        }
    });
env.execute();


// === EXAMPLE 3 ===

// convert to DataStream but write out the time attribute as a metadata column which means
// it is not part of the physical schema anymore

DataStream<Row> dataStream = tableEnv.toChangelogStream(
    table,
    Schema.newBuilder()
        .column("name", "STRING")
        .column("score", "INT")
        .columnByMetadata("rowtime", "TIMESTAMP_LTZ(3)")
        .build());

// the stream record's timestamp is defined by the metadata; it is not part of the Row

dataStream.process(
    new ProcessFunction<Row, Void>() {
        @Override
        public void processElement(Row row, Context ctx, Collector<Void> out) {

            // prints: [name, score]
            System.out.println(row.getFieldNames(true));

            // timestamp exists once
            System.out.println(ctx.timestamp());
        }
    });
env.execute();


// === EXAMPLE 4 ===

// for advanced users, it is also possible to use more internal data structures for efficiency

// note that this is only mentioned here for completeness because using internal data structures
// adds complexity and additional type handling

// however, converting a TIMESTAMP_LTZ column to `Long` or STRING to `byte[]` might be convenient,
// also structured types can be represented as `Row` if needed

DataStream<Row> dataStream = tableEnv.toChangelogStream(
    table,
    Schema.newBuilder()
        .column(
            "name",
            DataTypes.STRING().bridgedTo(StringData.class))
        .column(
            "score",
            DataTypes.INT())
        .column(
            "event_time",
            DataTypes.TIMESTAMP_LTZ(3).bridgedTo(Long.class))
        .build());

// leads to a stream of Row(name: StringData, score: Integer, event_time: Long)

将 Table API 管道添加到 DataStream API

一个 Flink 作业可以由多个彼此相邻运行的断开连接的管道组成。

在 Table API 中定义的 Source-to-sink 管道可以作为一个整体附加到，并且在调用DataStream APIStreamExecutionEnvironment 中的方法之一时将被提交。execute

但是，源不一定必须是表源，也可以是之前转换为 Table API 的另一个 DataStream 管道。因此，可以为 DataStream API 程序使用表接收器。

该功能可通过使用StreamStatementSet创建的专用实例获得 StreamTableEnvironment.createStatementSet()。通过使用语句集，计划者可以一起优化所有添加的语句，并提出一个或多个端到端管道，这些管道 StreamExecutionEnvironment在调用时添加StreamStatementSet.attachAsDataStream()。

以下示例显示了如何在一个作业中将表程序添加到 DataStream API 程序。

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.*;

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

StreamStatementSet statementSet = tableEnv.createStatementSet();

// create some source
TableDescriptor sourceDescriptor =
    TableDescriptor.forConnector("datagen")
        .option("number-of-rows", "3")
        .schema(
            Schema.newBuilder()
                .column("myCol", DataTypes.INT())
                .column("myOtherCol", DataTypes.BOOLEAN())
                .build())
        .build();

// create some sink
TableDescriptor sinkDescriptor = TableDescriptor.forConnector("print").build();

// add a pure Table API pipeline
Table tableFromSource = tableEnv.from(sourceDescriptor);
statementSet.add(tableFromSource.insertInto(sinkDescriptor));

// use table sinks for the DataStream API pipeline
DataStream<Integer> dataStream = env.fromElements(1, 2, 3);
Table tableFromStream = tableEnv.fromDataStream(dataStream);
statementSet.add(tableFromStream.insertInto(sinkDescriptor));

// attach both pipelines to StreamExecutionEnvironment
// (the statement set will be cleared after calling this method)
statementSet.attachAsDataStream();

// define other DataStream API parts
env.fromElements(4, 5, 6).addSink(new DiscardingSink<>());

// use DataStream API to submit the pipelines
env.execute();

// prints similar to:
// +I[1618440447, false]
// +I[1259693645, true]
// +I[158588930, false]
// +I[1]
// +I[2]
// +I[3]

TypeInformation 和 DataType 之间的映射

DataStream API 使用实例org.apache.flink.api.common.typeinfo.TypeInformation来描述在流中传输的记录类型。特别是，它定义了如何将记录从一个 DataStream 运算符序列化和反序列化到另一个。它还有助于将状态序列化为保存点和检查点。

Table API 使用自定义数据结构在内部表示记录，org.apache.flink.table.types.DataType 并向用户公开以声明将数据结构转换为的外部格式，以便在源、接收器、UDF 或 DataStream API 中更轻松地使用。

DataType比它更丰富，TypeInformation因为它还包括有关逻辑 SQL 类型的详细信息。因此，在转换过程中会隐式添加一些细节。

一个列名和类型Table自动从TypeInformation. DataStream用于DataStream.getType()检查是否已通过 DataStream API 的反射类型提取工具正确检测到类型信息。如果最外面的记录 TypeInformation是 一个 CompositeType，则在派生表的模式时它将在第一级展平。

DataStream API 并不总是能够TypeInformation根据反射提取更具体的内容。这通常会悄无声息地发生GenericTypeInfo，并得到通用 Kryo 序列化程序的支持。

例如，Row无法对类进行反射分析，并且始终需要明确的类型信息声明。如果 DataStream API 中没有声明正确的类型信息，则该行将显示为RAW 数据类型，并且 Table API 无法访问其字段。.map(…).returns(TypeInformation) 在 Java 或.map(…)(TypeInformation)Scala 中使用以显式声明类型信息。

TypeInformation 到 DataType

TypeInformation转换为 DataType时适用以下规则：

• 所有子类TypeInformation都映射到逻辑类型，包括与 Flink 内置序列化器对齐的可空性。

• 子类TupleTypeInfoBase被转换为行（for Row）或结构化类型（用于元组、POJO 和案例类）。

• BigDecimal默认转换为DECIMAL(38, 18)。

• 字段的顺序PojoTypeInfo由以所有字段作为参数的构造函数确定。如果在转换过程中未找到，则字段顺序将按字母顺序排列。

• GenericTypeInfo其他TypeInformation不能表示为所列之一的 org.apache.flink.table.api.DataTypes将被视为黑盒RAW类型。当前会话配置用于实现原始类型的序列化程序。届时将无法访问复合嵌套字段。

• 有关完整的翻译逻辑，请参阅 TypeInfoDataTypeConverter 。

用于DataTypes.of(TypeInformation)在自定义模式声明或 UDF 中调用上述逻辑。

DataType 到 TypeInformation

表运行时将确保正确地将输出记录序列化到 DataStream API 的第一个运算符。

之后，需要考虑 DataStream API 的类型信息语义。