【flink番外篇】16、DataStream 和 Table 相互转换示例

Flink 系列文章

一、Flink 专栏

Flink 专栏系统介绍某一知识点，并辅以具体的示例进行说明。

• 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。
• 2、Flink基础系列 本部分介绍Flink 的基础部分，比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。
• 3、Flik Table API和SQL基础系列 本部分介绍Flink Table Api和SQL的基本用法，比如Table API和SQL创建库、表用法、查询、窗口函数、catalog等等内容。
• 4、Flik Table API和SQL提高与应用系列 本部分是table api 和sql的应用部分，和实际的生产应用联系更为密切，以及有一定开发难度的内容。
• 5、Flink 监控系列 本部分和实际的运维、监控工作相关。

二、Flink 示例专栏

Flink 示例专栏是 Flink 专栏的辅助说明，一般不会介绍知识点的信息，更多的是提供一个一个可以具体使用的示例。本专栏不再分目录，通过链接即可看出介绍的内容。

两专栏的所有文章入口点击：Flink 系列文章汇总索引

文章目录

本文简单的介绍了DataStream 和 Table 的相互转换及示例。

如果需要了解更多内容，可以在本人Flink 专栏中了解更新系统的内容。

本文除了maven依赖外，没有其他依赖。

更多详细内容参考文章：

21、Flink 的table API与DataStream API 集成（完整版）

一、DataStream 和 Table 相互转换示例

1、maven依赖

<properties><encoding>UTF-8</encoding><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target><java.version>1.8</java.version><scala.version>2.12</scala.version><flink.version>1.17.0</flink.version></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-common</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java-bridge</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-sql-gateway</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-csv</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-planner --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_2.12</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java-uber --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java-uber</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-runtime</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc</artifactId><version>3.1.0-1.17</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.38</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-hive --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-hive_2.12</artifactId><version>1.17.0</version></dependency><dependency><groupId>org.apache.hive</groupId><artifactId>hive-exec</artifactId><version>3.1.2</version></dependency><!-- flink连接器 --><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>${flink.version}</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-sql-connector-kafka --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-sql-connector-kafka</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><!-- https://mvnrepository.com/artifact/org.apache.commons/commons-compress --><dependency><groupId>org.apache.commons</groupId><artifactId>commons-compress</artifactId><version>1.24.0</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.2</version><!-- <scope>provided</scope> --></dependency></dependencies>

2、 DataStream 和 Table 相互转换示例

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

1）、示例1 - toDataStream

下面的代码展示了如何在两个API之间来回切换的示例。

表的列名和类型自动从DataStream的TypeInformation派生。

由于DataStream API本机不支持变更日志处理，因此代码假设在流到表和表到流转换期间仅附加/仅插入语义。

importorg.apache.flink.streaming.api.datastream.DataStream;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.table.api.Table;importorg.apache.flink.table.api.bridge.java.StreamTableEnvironment;importorg.apache.flink.types.Row;/**
 * @author alanchan
 *
 */publicclassConvertingDataStreamAndTableDemo{/**
     * @param args
     * @throws Exception
     */publicstaticvoidmain(String[] args)throwsException{// 1、创建运行环境StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv =StreamTableEnvironment.create(env);// 2、创建输入流DataStream<String> dataStream = env.fromElements("alan","alanchan","alanchanchn");// 3、将datastream 转为 tableTable inputTable = tenv.fromDataStream(dataStream);// 4、创建视图，该步骤不是必须，将姓名转为大写
        tenv.createTemporaryView("InputTable", inputTable);Table resultTable = tenv.sqlQuery("SELECT UPPER(f0) FROM InputTable");// 5、将table转成datastream进行输出DataStream<Row> resultStream = tenv.toDataStream(resultTable);

        resultStream.print();
        env.execute();}}

• 示例输出

12>+I[ALAN]14>+I[ALANCHANCHN]13>+I[ALANCHAN]

fromDataStream和toDataStream的完整语义可以在下面的部分中找到。它还包括使用事件时间和水印。

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

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

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

2）、示例2 - toChangelogStream

下面的示例显示如何转换更新表。

每个结果行表示更改日志中的一个条目，该条目具有更改标志，可以通过对其调用row.getKind()来查询。

在本例中，alan的第二个分数在更改之前（-U）创建更新，在更改之后（+U）创建更新。

本示例仅仅以一个方法来展示，避免没有必要的代码，运行框架参考上述示例。

publicstaticvoidtest2()throwsException{// 1、创建运行环境StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv =StreamTableEnvironment.create(env);// 2、创建输入流DataStream<Row> dataStream = env.fromElements(Row.of("alan",18),Row.of("alanchan",19),Row.of("alanchanchn",20),Row.of("alan",20));// 3、将datastream 转为 tableTable inputTable = tenv.fromDataStream(dataStream).as("name","salary");// 4、创建视图，该步骤不是必须
        tenv.createTemporaryView("InputTable", inputTable);Table resultTable = tenv.sqlQuery("SELECT name, SUM(salary) FROM InputTable GROUP BY name");// 5、将table转成datastream进行输出DataStream<Row> resultStream = tenv.toChangelogStream(resultTable);

        resultStream.print();
        env.execute();}

• 运行结果

2>+I[alan,18]16>+I[alanchan,19]16>+I[alanchanchn,20]2>-U[alan,18]2>+U[alan,38]

fromChangelogStream和toChangelogStream的完整语义可以在下面的部分中找到。它包括使用事件时间和水印。它讨论了如何为输入和输出流声明主键和变更日志模式。

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

在批处理中，可以在连续的阶段中执行运算符，这些阶段在发出结果之前使用整个输入表。

例如，连接操作符可以在执行实际连接之前对两个有界输入进行排序（即排序合并连接算法），或者在使用另一个输入之前从一个输入构建哈希表（即哈希连接算法的构建/探测阶段）。

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

3）、示例3 - 通过仅切换标志来处理批处理和流数据

下面的示例说明了统一管道能够通过仅切换标志来处理批处理和流数据。

publicstaticvoidtest3()throwsException{// 1、创建运行环境StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);StreamTableEnvironment tenv =StreamTableEnvironment.create(env);// 2、创建输入流DataStream<Row> dataStream = env.fromElements(Row.of("alan",18),Row.of("alanchan",19),Row.of("alanchanchn",20),Row.of("alan",20));// 3、将datastream 转为 tableTable inputTable = tenv.fromDataStream(dataStream).as("name","salary");// 4、创建视图，该步骤不是必须
        tenv.createTemporaryView("InputTable", inputTable);Table resultTable = tenv.sqlQuery("SELECT name, SUM(salary) FROM InputTable GROUP BY name");// 5、将table转成datastream进行输出DataStream<Row> resultStream = tenv.toChangelogStream(resultTable);

        resultStream.print();
        
        env.execute();}

• 运行结果

注意比较和示例2的输出区别

+I[alanchan,19]+I[alan,38]+I[alanchanchn,20]

一旦将changelog 应用于外部系统（例如键值存储），可以看到两种模式都能够产生完全相同的输出表。通过在发出结果之前使用所有输入数据，批处理模式的更改日志仅由仅插入的更改组成。

以上，本文简单的介绍了DataStream 和 Table 的相互转换及示例。