FlinkSql之TableAPI详解 _生活百科

一、FlinkSql的概念核心概念Flink 的 Table API 和 SQL 是流批统一的 API 。这意味着 Table API & SQL 在无论有限的批式输入还是无限的流式输入下，都具有相同的语义。因为传统的关系代数以及 SQL 最开始都是为了批式处理而设计的，关系型查询在流式场景下不如在批式场景下容易理解.
动态表和连续查询动态表(Dynamic Tables) 是 Flink 的支持流数据的 Table API 和 SQL 的核心概念。
与表示批处理数据的静态表不同，动态表是随时间变化的。可以像查询静态批处理表一样查询它们。查询动态表将生成一个连续查询(Continuous Query) 。一个连续查询永远不会终止，结果会生成一个动态表。查询不断更新其(动态)结果表，以反映其(动态)输入表上的更改。
TableAPI
首先需要导入依赖
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-csv</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-json</artifactId> <version>${flink.version}</version> </dependency> ?  <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-compress</artifactId> <version>1.21</version> </dependency> /** * 使用TableAPI的基本套路： * 1.创建表的执行环境 * 2.创建表，将流转换为动态表，表的字段名从bean的属性名自动抽取 * 3.对动态表进行查询 * 4.把动态表转换为流 */这里需要注意的问题：
1.TableAPI 中将动态表转换为流时有两种方法
DataStream<Row> rowDataStream = tableEnvironment.toAppendStream(result, Row.class);toAppendStream方法只能在查询时使用，不能使用包含聚合函数等更新语句
DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnvironment.toRetractStream(select, Row.class);toRetractStream则可以使用
2.上述两种方法内传入的参数Row.class，表示将表中查询出的数据封装为行类型，也就是对每行进行封装，解决查询出的数据列少于或者多于原表。如何能够确保所查询的数据与之前封装的Bean有完全一致的结构则也可以封装为原Bean.class
代码实现：
package net.cyan.FlinkSql; ? import net.cyan.POJO.WaterSensor; import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; 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; ? import static org.apache.flink.table.api.Expressions.$; ? /** * 使用TableAPI的基本套路： * 1.创建表的执行环境 * 2.创建表，将流转换为动态表，表的字段名从bean的属性名自动抽取 * 3.对动态表进行查询 * 4.把动态表转换为流 */ public class Demo1 { public static void main(String[] args) { Configuration configuration=new Configuration(); configuration.setInteger("rest.port",3333); //创建执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration); env.setParallelism(1); //模拟数据 DataStreamSource<WaterSensor> WaterSensorSource = env.fromElements( new WaterSensor("S1", 1000L, 10), new WaterSensor("S1", 1000L, 10), new WaterSensor("S2", 2000L, 20), new WaterSensor("S3", 3000L, 30), new WaterSensor("S4", 4000L, 40), new WaterSensor("S5", 5000L, 50) ); //创建表的执行环境 StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env); //创建表，将流转换为动态表，表的字段名从bean的属性名自动抽取 Table table = tableEnvironment.fromDataStream(WaterSensorSource); //对表进行查询 //1、过时的查询书写 Table result = table .where("id='S1'") .select("*"); //2、不过时的书写 Table result1 = table // .where($("id").isEqual("S1")) .select($("id"), $("ts"), $("vc")); //3.聚合函数 Table select = table .groupBy($("id")) .aggregate($("vc").sum().as("sum_vc")) .select($("id"), $("sum_vc")); //把动态表转换为流，使用到了之前创建的表运行环境 ? SingleOutputStreamOperator<Row> tuple2DataStream = tableEnvironment .toRetractStream(select, Row.class) .filter(t -> t.f0) .map(t -> t.f1); // DataStream<Row> rowDataStream = tableEnvironment.toAppendStream(result, Row.class); // DataStream<Row> rowDataStream1 = tableEnvironment.toAppendStream(result1, Row.class); // rowDataStream.print(); // rowDataStream1.print(); tuple2DataStream.print(); ? ? try { //启动执行环境 env.execute(); } catch (Exception e) { e.printStackTrace(); } ? ? ? } }

FlinkSql之TableAPI详解

推荐阅读

不倒翁小鸟的制作教程

山药圆子怎么做才好吃，肉末山药圆子的做法？

熟黄芪泡水喝的功效与作用熟黄芪泡水喝有哪些功效与作用

饭圈拉瓜是什么意思饭圈拉郎是什么意思

父母买房可以写未成年孩子的名字吗,规避转让税但不能贷款

关于特别认真地生活简述特别认真地生活

办银行卡需要什么办银行卡需要什么证件和材料

关于淮北公交简述淮北公交

大红袍分几种肉桂

孕妇铅过高有什么症状有什么不良影响

终不似少年游终不似少年游

柠檬片的功能与功效柠檬片的功效作用

防弹咖啡怎么做好喝防弹咖啡怎么做

excel查找不出来是什么原因

今年广东调整养老金的新消息 2022广东省养老金上调方案及补发时间

毓字取名女孩有寓意属兔最搭的名字有哪些？

强势老人是家庭的灾难【老人强势发生家庭矛盾】

昆明机场离哪个高铁站近，昆明飞机场到高铁站有多远？

“魑魅魍魉” 你们知道怎么读吗？魑魅魍魉怎么读

糖尿病人适合送的礼品女总裁给男秘书送粥