「我的数据空间」实时计算实践笔记 · Flink SQL 系列

使用 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的常用处理方式:

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;

本文收录于「我的数据空间」技术库——一套可私有化部署的数据平台(数据集成 / 实时计算 / 数据湖 / 湖仓查询 / 智能问数)。