FlinkSQL写入JDBC数据支持ignoreDelete

使用场景

在FlinkSQL作业中，双流Join或聚合操作会产生回撤消息（如DELETE和UPDATE_BEFORE），当这些消息写入JDBC目标表时，可能导致数据被误删或覆盖。FlinkSQL写入JDBC数据时支持通过记录过滤功能过滤掉DELETE和UPDATE_BEFORE状态的数据。

约束与限制

本章节仅适用于MRS 3.3.0-LTS及之后版本。

使用方法

在创建的JDBC Connector Sink流表中添加“filter.record.enabled”和“filter.row-kinds”参数。

• filter.record.enabled：该参数用于控制是否启用记录过滤功能。配置为“true”时启用记录过滤功能，Flink会根据指定的过滤条件过滤记录。配置为“false”或不配置时，不启用记录过滤功能。

  当需要过滤JDBC Sink中特定类型的记录时，需设置此参数为“true”。

• filter.row-kinds：该参数用于指定需要过滤的行类型。当“filter.record.enabled”为“true”时，需设置此参数指定要过滤的行类型，多个行类型用逗号分隔。
  • INSERT：插入记录，一般不需要过滤。
  • UPDATE_AFTER：更新后的新记录，一般不需要过滤。
  • UPDATE_BEFORE：更新前的旧记录，常用于过滤双流Join产生的回撤数据。
  • DELETE：删除记录，常用于过滤不需要删除操作的场景。

SQL示例如下：

CREATE TABLE user_score (
  idx varchar(20),
  user_id varchar(20),
  score bigint
) WITH (
  'connector' = 'kafka',
  'topic' = 'topic-qk',
  'properties.bootstrap.servers' = 'xxxx:21005',
  'properties.group.id' = 'test_qk',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'csv'
);
CREATE TABLE mysql_output (
  idx varchar(20),
  user_id varchar(20),
  all_score bigint,
  PRIMARY KEY(idx, user_id) NOT ENFORCED
) WITH(
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://MySQL的服务器IP:MySQL的服务器端口/mysql',
  'table-name' = 'customer_t1',
  'username' = 'username', --连接MySQL数据库的用户名
  'password' = 'password',--连接MySQL数据库的密码
  'filter.record.enabled' = 'true', --表示启用记录过滤功能
  'filter.row-kinds' = 'UPDATE_BEFORE' --表示在写入数据时，过滤掉UPDATE_BEFORE类型的记录
);
insert into
  mysql_output
select
  idx,
  user_id,
  sum(score) as all_score
from
  user_score
group by
  idx,
  user_id;