上接:flink1.12.x Table API 和 SQL:Table 转换成 DataStream
说明
有些流式查询,其结果是动态更新的,如:聚合查询
select count(1) from table
- 来第一条数据,查询结果是1
- 来第二条数据,查询结果是2,应该删掉上一条结果
由更新模式(Update Mode)实现上面的功能,有以下3种模式
- Append
- Retract
- Upsert
追加(Append)模式
表只做追加操作,比如:普通 select 查询,如下:
select * from t_order where price>60
撤回(Retract)模式(常用)
表和外部连接器交换添加(Add)和撤回(Retract)消息
- 插入操作(Insert)编码为
Add消息; - 删除(Delete)编码为
Retract消息; - 更新(Update)编码为:上一条的
Retract和下一条的Add消息
应用场景
任何场景均适合,如:聚合查询、删除、更新
例子
flink1.12.x Table API 和 SQL:从socket读取数据,实现分组聚合查询(Retract)
flink1.12.x Table API 和 SQL:读取csv文件,实现分组聚合查询
显示结果如下:
2> (true,lucy,120.0)
4> (true,韩梅梅,90.0)
2> (true,李雷,80.0)
2> (false,李雷,80.0)
2> (true,李雷,90.0)
4> (false,韩梅梅,90.0)
4> (true,韩梅梅,80.0)
2> (false,lucy,120.0)
2> (true,lucy,110.0)
true 表示 add 操作false 表示 Retract 操作(收回、删除)
更新插入(Upsert)模式
- 更新和插入都被编码为 Upsert 消息
- 删除编码为 Delete 消息
需要指定 主键,如:执行聚合查询,并将结果保存到 elasticsearch 中
参考:
https://cloud.tencent.com/developer/article/1483316

