flinkSQL Table转DataStream

flink版本1.14

flinksql 来源于kafka json格式数据

变化的表

业务中sql可能不完全满足使用，需要转换成DataStream 更灵活一些，所以需要互相转换，发挥各自的优势。

         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(10000);env.setParallelism(1);final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);//两个来自kafka的表tEnv.executeSql(CreateTableSQL.kafkaTableInfo);tEnv.executeSql(CreateTableSQL.kafkaTablePerson);//join两个表Table tableResult = tEnv.sqlQuery(SelectSQL.selectPerMaxScore);tEnv.toDataStream(tableResult, Row.class);dataStream.flatMap(new FlatMapFunction<Row, Object>() {@Overridepublic void flatMap(Row value, Collector<Object> out) throws Exception {String s = value.getField(0) + ":" + value.getField(1);out.collect(s);}}).print();env.execute();

会发现报错

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

那是因为在table-to-stream中，数据在发生变化，因此需要用toChangelogStream来转换

修改成如下内容：

         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(10000);env.setParallelism(1);final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);//两个来自kafka的表tEnv.executeSql(CreateTableSQL.kafkaTableInfo);tEnv.executeSql(CreateTableSQL.kafkaTablePerson);//join两个表Table tableResult = tEnv.sqlQuery(SelectSQL.selectPerMaxScore);tEnv.toChangelogStream(tableResult).flatMap(new FlatMapFunction<Row, Object>() {@Overridepublic void flatMap(Row value, Collector<Object> out) throws Exception {String s = value.getField(0) + ":" + value.getField(1);out.collect(s);}}).print();env.execute();

在网上找了一圈，没有类似的文章，自己做个记录。
这里虽然也可以用createTemporaryView来把表转换成DataStream，但是这种方式的表是固定的，在实际应用的使用场景中，还是toChangelogStream的应用更广一些。

executeSql() 与 sqlQuery()

你可能会发现有的地方用的是executeSql() 有的地方用的是 sqlQuery() ,这两者是什么不同呢，对此我特意去看了一下源码里的注释。

sqlQuery(String query)

针对sqlQuery()他是这么说的

评估对已注册表的 SQL 查询并返回一个 Table 对象，该对象描述了进一步转换的管道。查询引用的所有表和其他对象都必须在 TableEnvironment 中注册。例如，使用 createTemporaryView(String, Table)) 来引用 Table 对象或使用 createTemporarySystemFunction(String, Class) 来引用函数。
或者，当调用 Table.toString() 方法时，会自动注册 Table 对象，例如当它被嵌入到字符串中时。因此，SQL 查询可以直接内联（即匿名）引用 Table 对象，如下所示：
Table table = ...;
String tableName = table.toString();
// 表没有注册到表环境
tEnv.sqlQuery("SELECT * FROM " + tableName + " WHERE a > 12");
请注意，返回的 Table 是一个 API 对象，仅包含管道描述。它实际上对应于 SQL 术语中的视图。调用 Table.execute() 触发执行或直接使用 executeSql(String)。

executeSql(String statement)

执行给定的单个语句并返回执行结果。
语句可以是 DDL/DML/DQL/SHOW/DESCRIBE/EXPLAIN/USE。对于 DML 和 DQL，此方法在提交作业后返回 TableResult。对于 DDL 和 DCL 语句，操作完成后将返回 TableResult。
如果多个管道应将数据作为单个执行的一部分插入到一个或多个接收器表中，请使用 StatementSet（请参阅 createStatementSet()）。
默认情况下，所有 DML 操作都是异步执行的。使用 TableResult.await() 或 TableResult.getJobClient() 来监控执行。设置 TableConfigOptions.TABLE_DML_SYNC 以始终同步执行。

需要强调的是，这两个是不能强行互相转换的，出现以下报错

java.lang.ClassCastException:org.apache.flink.table.api.internal.TableResultImpl cannot be cast to org.apache.flink.table.api.Table

还有一点， sqlQuery(String query)不能执行复杂的语句，会出现以下报错

Unsupported SQL query! sqlQuery() only accepts a single SQL query of type SELECT, UNION, INTERSECT, EXCEPT, VALUES, and ORDER_BY.

固定的表

如果需求只是一个固定的表可以通过这种下面的案例

     final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(10000);env.setParallelism(1);final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);// connector doris的一个表 PerCountNametEnv.executeSql(CreateTableSQL.PerCountNDoris);Table perCountName = tEnv.from("PerCountName");// PerCountName表对应的实体类 PerCountName.classtEnv.toDataStream(perCountName, PerCountName.class).flatMap(new FlatMapFunction<PerCountName, Object>() {@Overridepublic void flatMap(PerCountName value, Collector<Object> out) throws Exception {String s = value.getName()+"->"+value.getNum();out.collect(s);}}).print();env.execute();

flinkSQL Table转DataStream相关推荐

Flink Table 和 DataStream 转换
文章目录 Flink Table 和 DataStream 转换 1. 表(Table) 转换为流(DataStream) 1.1 处理(仅插入)流 1.1.1 fromDataStream()方法 ...
（十八）Flink Table API SQL 编程指南 Table API 和Datastream API 集成
文章目录 DataStream 和 Table 之间的转换依赖项和导入配置执行行为 datastream API table API 批处理运行时模式 Changelog统一处理(仅插入)流 ...
干货 | 五千字长文带你快速入门FlinkSQL
本文已收录github:https://github.com/BigDataScholar/TheKingOfBigData,里面有大数据高频考点,Java一线大厂面试题资源,上百本免费电子书籍,作者 ...
Flink的Table API 与SQL介绍及调用
1 概述 DataSetAPI和DateStreamAPI是基于整个Flink的运行时环境做操作处理的,Table API和SQL是在DateStreamAPI上又包了一层.对于新版本的Blin ...
Flink入门第十二课：DataStream api/Flink sql实现每隔5分钟统计最近一小时热门商品小案例
用到的数据文件用到的数据文件链接:https://pan.baidu.com/s/1uCk-IF4wWVfUkuuTAKaD0w 提取码:2hmu 1.需求 & 数据用户行为数据不断写入 ...
AppendStream和RetractStream(没有弄完)
Reference [1]Flink RetractStream示例及UDF函数实现 [2]flink实战--flinkSQL 追加模式与缩进模式(toRetractStream)的区别 [3]Fli ...
95-910-140-源码-FlinkSQL-FlinkSQL简介
文章目录 1.世界 2.概述 3.FlinkSQL执行计划 4.Table与DataStream和DataSet API集成 5.**Scala的隐式转换** 6.将DataStream或DataSe ...
Flink学习4-流式SQL
Flink学习4-流式SQL Flink系列文章更多Flink系列文章请点击Flink系列文章更多大数据文章请点击大数据好文推荐摘要介绍Flink Table Sql API相关概念,还会提供 ...
蝉联 Apache 最活跃项目，Flink 社区是如何保持高速发展的？
简介:2020 年是 Apache Flink 社区生态加速繁荣的一年. 本文由 Apache Flink 中文社区发起人,阿里云计算平台事业部实时计算与开放平台部门负责人王峰分享,主要介绍 Flin ...