我的数据空间

主题

Flink SQL 增量同步到 MySQL:自增主键、结果集过大与主键冲突 ​

将 Iceberg/Hive 数据增量同步至 MySQL

无自增主键场景 ​

sql
merge into
    mysql_10007_sample_database.sample_db.mysql_sample_table target
using(
    select
      channel as channel,
      -- 由于目标表的字段类型是BIGINT，因此需要转换字段类型。
      -- 注：根据自己的需求做强制类型转换
      cast(goods_id as BIGINT) as goods_id,
      cast(stock_num as BIGINT) as stock_num
    from
      hive_catalog.tmp.sample_table
    where
      day = ${date -1};
) source
-- 匹配条件
on target.channel = source.channel AND target.goods_id = source.goods_id
-- 如果匹配上，则做数据更新
when matched then update set
    target.stock_num = source.stock_num 
    
-- 如果匹配不上，则做插入
when not matched then insert *
;

存在自增主键 ​

如果存在自增主键， Flink SQL 会将自增主键作为主键， 如果用 0 作为新写入的默认值， 写入 MySQL 之前 Flink Jdbc connector 会将具有相同主键的数据合并，导致丢数， 因此如果存在自增主键，在 Flink SQL 侧应该忽略该主键并声明新的主键。

2.1 使用 CREATE TABLE LIKE 声明新的主键 ​

sql
-- 使用 create table like 声明新的主键

-- channel, goods_id 要对应 MySQL 的联合 unique_key
create table mysql_sink (primary key (channel, goods_id) not enforced) 
like mysql_10007_sample_database.sample_db.mysql_sample_table (excluding CONSTRAINTS); 

merge into
    mysql_sink target
using(
    select
      -- 这里先将id赋值为0
      cast(0 as bigint) as id,
      channel as channel,
      -- 由于目标表的字段类型是BIGINT，因此需要转换字段类型。
      -- 注：根据自己的需求做强制类型转换
      cast(goods_id as BIGINT) as goods_id,
      cast(stock_num as BIGINT) as stock_num
    from
      hive_catalog.tmp.sample_table
    where
      day = ${date -1};
) source
-- 匹配条件
on target.channel = source.channel AND target.goods_id = source.goods_id
-- 如果匹配上，则做数据更新
when matched then update set
    target.stock_num = source.stock_num 
    
-- 如果匹配不上，则做插入
when not matched then insert *
;

2.2 设置发送 batch 数为 1 (可避免写入前基于主键合并) ​

通过 hint 设置参数， 禁用发送前的攒批操作， 避免合并数据， 但存在效率问题

sql

merge into
    mysql_10007_sample_database.sample_db.mysql_sample_table 
         /*+ OPTIONS('sink.buffer-flush.max-rows'='1') */
    target
using(
    select
      -- 这里先将id赋值为0
      cast(0 as bigint) as id,
      channel as channel,
      -- 由于目标表的字段类型是BIGINT，因此需要转换字段类型。
      -- 注：根据自己的需求做强制类型转换
      cast(goods_id as BIGINT) as goods_id,
      cast(stock_num as BIGINT) as stock_num
    from
      hive_catalog.tmp.sample_table
    where
      day = ${date -1};
) source
-- 匹配条件
on target.channel = source.channel AND target.goods_id = source.goods_id
-- 如果匹配上，则做数据更新
when matched then update set
    target.stock_num = source.stock_num 
    
-- 如果匹配不上，则做插入
when not matched then insert *
;

使用FlinkSQL完成无效数据清理 ​

希望从目标表中删除源表中不存在记录，如下所示：

sql
merge into mysql_10007_sample_database.sample_db.mysql_sample_table target
using (
    -- 构造出一份全集合数据，源表中不存在的记录标记为删除
    select
        -- 确保能够获取到channel
        coalesce(t1.channel, t2.channel) as channel,
        -- 增加强制类型转换，确保类型匹配
        cast(coalesce(t1.goods_id, t2.goods_id) as BIGINT) as goods_id
        cast(t1.stock_num as BIGINT) as stock_num
        ,coalesce(t1.goods_id, 'delete') as mark --删除mysql标识
    from hive_catalog.tmp.sample_table t1
    full join mysql_10007_sample_database.sample_db.mysql_sample_table t2
        on t1.channel = t2.channel AND t1.goods_id = t2.goods_id
    where t1.day = ${date-1}
) source
-- 匹配条件
on target.channel = source.channel AND target.goods_id = source.goods_id
-- 如果匹配上，且标记为删除，则做删除
when matched and source.mark = 'delete' then delete
-- 如果匹配上，则做数据更新
when matched then update set
    target.stock_num = source.stock_num 
-- 如果匹配不上，则做插入
when not matched then insert *
;

要点：

  • 构造出一份全集合数据，源表中不存在的记录标记为删除
  • 利用merge into + 删除标记，做删除

常见问题 ​

4.1 读 JDBC 数据量太大，报错 Query result set is too large ​

解决方案： 增加分区读参数， 减少单次读 MySQL 的数据量

  • 使用 hint 增加分区读的参数， 分别指定分区字段(scan.partition.column)， 分区的上界(scan.partition.lower-bound)，分区的下界(scan.partition.upper-bound)，以及分区数(scan.partition.num)
  • 分区字段一定要为数值类型
sql
merge into
    mysql_10007_sample_database.sample_db.mysql_sample_table 
     /*+ OPTIONS('scan.partition.column'='id', 'scan.partition.num' = '20000', 'scan.partition.lower-bound' = '0', 'scan.partition.upper-bound' = '2000000000') */
    target
using(
)

4.2 不支持多条 delete 语句 ​

sql
: unknown error: client's Capabilities not support multi statements,but proxy receive multi statements:
  1. 修改为单条发送:
sql
merge into
    mysql_10007_sample_database.sample_db.mysql_sample_table 
     /*+ OPTIONS('sink.buffer-flush.max-rows' = '1') */
    target
using(
)
  1. 如果使用临时表指定 Jdbc Url，需要在 jdbc URL 中增加 allowMultiQueries=true
java
?allowMultiQueries=true
  1. 若链路上有 MySQL 代理，需要由 DBA 侧放开多语句支持（部分代理默认关闭）

4.3 写入 mysql 主键冲突问题 ​

元数据服务未传入 mysql 主键信息导致 merge into 语句中的 update 语句的写入变成了 insert or fail，需要自己通过 create table like 声明主键。

java
CREATE TABLE targetMysql (
    `req_id` VARCHAR NOT NULL,
    `user_email` VARCHAR NOT NULL,
    PRIMARY KEY (`req_id`, `user_email`) NOT ENFORCED --指定主键
) WITH (
)
LIKE mysql_10033_sample_database.sample_db.sample_project
(OVERWRITING PHYSICAL)
;

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