「我的数据空间」实时计算实践笔记 · Flink SQL 系列
使用 Flink 临时表
使用 DDL 声明 对应的 schema 和 format
CREATETABLEKafkaSource(idVARCHAR,`count`BIGINT,changelogBOOLEAN)with('topic'='topic_ods_order_event','connector'='kafka','format'='binlog','binlog.with-changelog'='true','properties.bootstrap.servers'='kafka-bootstrap:9092','scan.startup.mode'='latest-offset');说明:
- changelog 是一个固定的字段名， 用于表示消息的属性, true代表增加， false代表删除, changelog必须在 ‘connector.with-changelog’ = 'true’时才会生效， 否则，chanlog会被当作一个普通字段， 如果原始的mysql表中不包含这个changelog字段，则会报错.
- 如果原始表中已经存在了changelog这个字段，且设置了changelog字段， changelog字段会优先作为消息的属性信息，而不是原始的字段， 为避免冲突，可以设置 ‘connector.changelog-name’ = ‘xxx’ 来修改用于存放changelog的字段名.
除去changelog之外，还支持获得binlog的其他属性
| 字段名 | 说明 | 配置参数 | 配置字段名 |
|---|---|---|---|
| changelog | binlog的消息属性 | ‘connector.with-changelog’ | ‘connector.changelog-name’ |
| offset | binlog的offset | ‘connector.with-offset’ | ‘connector.offset-name’ |
| binlogTime | binlog产生的时间 | ‘connector.with-timestamp’ | ‘connector.timestamp-name’ |
changelog的常用处理方式:
3.1 直接过滤掉删除的记录
SELECT*FROMKafkaSourceWHEREchangelog=true;这种情况只适合于没有主键删除的情况， 只需要处理add的消息即可.
3.2 将changelog字段作为普通字段处理
SELECTid,LAST_VALUE(`count`),LAST_VALUE(changelog)FROMKafkaSourceGROUPBYid;在FlinkSQL这一层不处理changelog，而是将changelog当作普通字段来处理， 并写入到下游系统， 由下游的系统来处理.
原始数据:
+0001,1,true+0001,2,true+0001,2,false经过处理之后
+0001,1,true-0001,2,true+0001,2,true-0001,2,false+0001,2,false经过以上的处理， 最终 + 0001， 2， false这条记录会被写入到最终的sink表.
典型应用场景:
binlog数据实时导入到iceberg中:
CREATETABLEKafkaSource(idVARCHAR,`count`BIGINT,changelogBOOLEAN)with('topic'='topic_ods_order_event','connector'='kafka','format'='binlog','binlog.with-changelog'='true','properties.bootstrap.servers'='kafka-bootstrap:9092','scan.startup.mode'='latest-offset');insertintoiceberg_catalog.dw.dwd_orderselectevent_guidasguid,event_typeastypefromKafkaSourcegroupbyevent_guid,event_type;本文收录于「我的数据空间」技术库——一套可私有化部署的数据平台(数据集成 / 实时计算 / 数据湖 / 湖仓查询 / 智能问数)。
