FlinkSQL 处理 binlog:changelog 的三种处理方式

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