「我的数据空间」实时计算实践笔记 · Flink SQL 系列将 Iceberg/Hive 数据增量同步至 MySQL无自增主键场景mergeintomysql_10007_sample_database.sample_db.mysql_sample_table targetusing(selectchannelaschannel,-- 由于目标表的字段类型是BIGINT因此需要转换字段类型。-- 注根据自己的需求做强制类型转换cast(goods_idasBIGINT)asgoods_id,cast(stock_numasBIGINT)asstock_numfromhive_catalog.tmp.sample_tablewhereday${date-1};)source-- 匹配条件ontarget.channelsource.channelANDtarget.goods_idsource.goods_id-- 如果匹配上则做数据更新whenmatchedthenupdatesettarget.stock_numsource.stock_num-- 如果匹配不上则做插入whennotmatchedtheninsert*;存在自增主键如果存在自增主键 Flink SQL 会将自增主键作为主键 如果用 0 作为新写入的默认值 写入 MySQL 之前 Flink Jdbc connector 会将具有相同主键的数据合并导致丢数 因此如果存在自增主键在 Flink SQL 侧应该忽略该主键并声明新的主键。2.1 使用 CREATE TABLE LIKE 声明新的主键-- 使用 create table like 声明新的主键-- channel, goods_id 要对应 MySQL 的联合 unique_keycreatetablemysql_sink(primarykey(channel,goods_id)notenforced)likemysql_10007_sample_database.sample_db.mysql_sample_table(excluding CONSTRAINTS);mergeintomysql_sink targetusing(select-- 这里先将id赋值为0cast(0asbigint)asid,channelaschannel,-- 由于目标表的字段类型是BIGINT因此需要转换字段类型。-- 注根据自己的需求做强制类型转换cast(goods_idasBIGINT)asgoods_id,cast(stock_numasBIGINT)asstock_numfromhive_catalog.tmp.sample_tablewhereday${date-1};)source-- 匹配条件ontarget.channelsource.channelANDtarget.goods_idsource.goods_id-- 如果匹配上则做数据更新whenmatchedthenupdatesettarget.stock_numsource.stock_num-- 如果匹配不上则做插入whennotmatchedtheninsert*;2.2 设置发送 batch 数为 1 (可避免写入前基于主键合并)通过 hint 设置参数 禁用发送前的攒批操作 避免合并数据但存在效率问题mergeintomysql_10007_sample_database.sample_db.mysql_sample_table/* OPTIONS(sink.buffer-flush.max-rows1) */targetusing(select-- 这里先将id赋值为0cast(0asbigint)asid,channelaschannel,-- 由于目标表的字段类型是BIGINT因此需要转换字段类型。-- 注根据自己的需求做强制类型转换cast(goods_idasBIGINT)asgoods_id,cast(stock_numasBIGINT)asstock_numfromhive_catalog.tmp.sample_tablewhereday${date-1};)source-- 匹配条件ontarget.channelsource.channelANDtarget.goods_idsource.goods_id-- 如果匹配上则做数据更新whenmatchedthenupdatesettarget.stock_numsource.stock_num-- 如果匹配不上则做插入whennotmatchedtheninsert*;使用FlinkSQL完成无效数据清理希望从目标表中删除源表中不存在记录如下所示mergeintomysql_10007_sample_database.sample_db.mysql_sample_table targetusing(-- 构造出一份全集合数据源表中不存在的记录标记为删除select-- 确保能够获取到channelcoalesce(t1.channel,t2.channel)aschannel,-- 增加强制类型转换确保类型匹配cast(coalesce(t1.goods_id,t2.goods_id)asBIGINT)asgoods_id cast(t1.stock_numasBIGINT)asstock_num,coalesce(t1.goods_id,delete)asmark--删除mysql标识fromhive_catalog.tmp.sample_table t1fulljoinmysql_10007_sample_database.sample_db.mysql_sample_table t2ont1.channelt2.channelANDt1.goods_idt2.goods_idwheret1.day${date-1})source-- 匹配条件ontarget.channelsource.channelANDtarget.goods_idsource.goods_id-- 如果匹配上且标记为删除则做删除whenmatchedandsource.markdeletethendelete-- 如果匹配上则做数据更新whenmatchedthenupdatesettarget.stock_numsource.stock_num-- 如果匹配不上则做插入whennotmatchedtheninsert*;要点构造出一份全集合数据源表中不存在的记录标记为删除利用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)分区字段一定要为数值类型mergeintomysql_10007_sample_database.sample_db.mysql_sample_table/* OPTIONS(scan.partition.columnid, scan.partition.num 20000, scan.partition.lower-bound 0, scan.partition.upper-bound 2000000000) */targetusing()4.2 不支持多条 delete 语句: unknown error: clients Capabilitiesnotsupport multi statements,but proxy receive multi statements:修改为单条发送:mergeintomysql_10007_sample_database.sample_db.mysql_sample_table/* OPTIONS(sink.buffer-flush.max-rows 1) */targetusing()如果使用临时表指定 Jdbc Url需要在 jdbc URL 中增加 allowMultiQueriestrue?allowMultiQueriestrue若链路上有 MySQL 代理需要由 DBA 侧放开多语句支持部分代理默认关闭4.3 写入 mysql 主键冲突问题元数据服务未传入 mysql 主键信息导致 merge into 语句中的 update 语句的写入变成了 insert or fail需要自己通过 create table like 声明主键。CREATETABLEtargetMysql(req_idVARCHARNOTNULL,user_emailVARCHARNOTNULL,PRIMARYKEY(req_id,user_email)NOTENFORCED--指定主键)WITH()LIKEmysql_10033_sample_database.sample_db.sample_project(OVERWRITINGPHYSICAL);本文收录于「我的数据空间」技术库——一套可私有化部署的数据平台(数据集成 / 实时计算 / 数据湖 / 湖仓查询 / 智能问数)。产品介绍见我的数据空间官网。
