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

使用 Flink 临时表

使用 DDL 声明 对应的 schema 和 format

CREATE TABLE KafkaSource (
  id VARCHAR,
  `count` BIGINT,
  changelog  BOOLEAN
) 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'
);

说明:

1. changelog 是一个固定的字段名， 用于表示消息的属性, true代表增加， false代表删除, changelog必须在 'connector.with-changelog' = 'true'时才会生效， 否则，chanlog会被当作一个普通字段， 如果原始的mysql表中不包含这个changelog字段，则会报错.
2. 如果原始表中已经存在了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 * FROM KafkaSource
WHERE changelog = true;

这种情况只适合于没有主键删除的情况， 只需要处理add的消息即可.

3.2 将changelog字段作为普通字段处理

SELECT id, LAST_VALUE(`count`), LAST_VALUE(changelog) 
FROM KafkaSource
GROUP BY id;

在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中:

CREATE TABLE KafkaSource (
  id VARCHAR,
  `count` BIGINT,
  changelog  BOOLEAN
) 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'
);


insert into iceberg_catalog.dw.dwd_order 
select event_guid as guid, event_type as type
from KafkaSource
group by event_guid, event_type;

---

本文是「我的数据空间」实时计算实践笔记的一篇。平台把这类链路做成了托管作业(Flink SQL 流作业、实时集成、Iceberg 表管理与维护),支持私有化部署与 OEM 合作 —— 见核心能力总览与技术博客,交流合作 QQ:1559851993。