惠阳市网站建设,的网站开发工具有哪些,wordpress 点击加微信,上海网站建设推广Flink 系列文章
一、Flink 专栏
Flink 专栏系统介绍某一知识点#xff0c;并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分#xff0c;比如术语、架构、编程模型、编程指南、基本的…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 系列文章汇总索引 文章目录 Flink 系列文章一、DataStream 和 Table 相互转换示例1、maven依赖2、 DataStream 和 Table 相互转换示例1、示例1 - toDataStream2、示例2 - toChangelogStream3、示例3 - 通过仅切换标志来处理批处理和流数据 本文简单的介绍了DataStream 和 Table 的相互转换及示例。
如果需要了解更多内容可以在本人Flink 专栏中了解更新系统的内容。
本文除了maven依赖外没有其他依赖。
更多详细内容参考文章
21、Flink 的table API与DataStream API 集成完整版
一、DataStream 和 Table 相互转换示例
1、maven依赖
propertiesencodingUTF-8/encodingproject.build.sourceEncodingUTF-8/project.build.sourceEncodingmaven.compiler.source1.8/maven.compiler.sourcemaven.compiler.target1.8/maven.compiler.targetjava.version1.8/java.versionscala.version2.12/scala.versionflink.version1.17.0/flink.version/propertiesdependenciesdependencygroupIdorg.apache.flink/groupIdartifactIdflink-clients/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-java/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-common/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-java/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-api-java-bridge/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-sql-gateway/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-csv/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-json/artifactIdversion${flink.version}/versionscopeprovided/scope/dependency!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-planner --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-planner_2.12/artifactIdversion${flink.version}/versionscopeprovided/scope/dependency!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java-uber --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-api-java-uber/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-runtime/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-jdbc/artifactIdversion3.1.0-1.17/version/dependencydependencygroupIdmysql/groupIdartifactIdmysql-connector-java/artifactIdversion5.1.38/version/dependency!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-hive --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-hive_2.12/artifactIdversion1.17.0/version/dependencydependencygroupIdorg.apache.hive/groupIdartifactIdhive-exec/artifactIdversion3.1.2/version/dependency!-- flink连接器 --!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-kafka/artifactIdversion${flink.version}/version/dependency!-- https://mvnrepository.com/artifact/org.apache.flink/flink-sql-connector-kafka --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-sql-connector-kafka/artifactIdversion${flink.version}/versionscopeprovided/scope/dependency!-- https://mvnrepository.com/artifact/org.apache.commons/commons-compress --dependencygroupIdorg.apache.commons/groupIdartifactIdcommons-compress/artifactIdversion1.24.0/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.2/version!-- scopeprovided/scope --/dependency/dependencies2、 DataStream 和 Table 相互转换示例
Flink提供了专门的StreamTableEnvironment用于与DataStream API集成。这些环境使用其他方法扩展常规TableEnvironment并将DataStream API中使用的StreamExecutionEnvironments作为参数。
1、示例1 - toDataStream
下面的代码展示了如何在两个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;/*** author alanchan**/
public class ConvertingDataStreamAndTableDemo {/*** param args* throws Exception*/public static void main(String[] args) throws Exception {// 1、创建运行环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv StreamTableEnvironment.create(env);// 2、创建输入流DataStreamString 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进行输出DataStreamRow 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 doesnt support consuming update changes [...].在这种情况下需要再次修改查询或切换到ChangelogStream。
2、示例2 - toChangelogStream
下面的示例显示如何转换更新表。
每个结果行表示更改日志中的一个条目该条目具有更改标志可以通过对其调用row.getKind()来查询。
在本例中alan的第二个分数在更改之前-U创建更新在更改之后U创建更新。
本示例仅仅以一个方法来展示避免没有必要的代码运行框架参考上述示例。 public static void test2() throws Exception {// 1、创建运行环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv StreamTableEnvironment.create(env);// 2、创建输入流DataStreamRow 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进行输出DataStreamRow 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 - 通过仅切换标志来处理批处理和流数据
下面的示例说明了统一管道能够通过仅切换标志来处理批处理和流数据。
public static void test3() throws Exception {// 1、创建运行环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.BATCH);StreamTableEnvironment tenv StreamTableEnvironment.create(env);// 2、创建输入流DataStreamRow 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进行输出DataStreamRow resultStream tenv.toChangelogStream(resultTable);resultStream.print();env.execute();}运行结果
注意比较和示例2的输出区别
I[alanchan, 19]
I[alan, 38]
I[alanchanchn, 20]
一旦将changelog 应用于外部系统例如键值存储可以看到两种模式都能够产生完全相同的输出表。通过在发出结果之前使用所有输入数据批处理模式的更改日志仅由仅插入的更改组成。
以上本文简单的介绍了DataStream 和 Table 的相互转换及示例。
