使用Flink CDC同步MySQL到Iceberg时删改操作引发重复行问题

我创建了如下Iceberg表：

CREATE TABLE ars.results(
  "id" varchar, 
  "input_md5" varchar, 
  "output_md5" varchar, 
  "log" varchar, 
  "metric" varchar, 
  "create_time" TIMESTAMP(6) WITH TIME ZONE, 
  "update_time" TIMESTAMP(6) WITH TIME ZONE, 
  "workflow_id_id" varchar, 
  "error_details" varchar, 
  "error_stage" varchar, 
  "error_type" varchar
)
WITH ( format = 'PARQUET',partitioning = ARRAY['hour(create_time)'])

随后使用Flink CDC将MySQL数据同步至该Iceberg表，代码如下：

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment
from pyflink.datastream.checkpointing_mode import CheckpointingMode

def run():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1).enable_checkpointing(2000, CheckpointingMode.EXACTLY_ONCE)
    
    table_env = StreamTableEnvironment.create(env)

    table_env.execute_sql("""
        CREATE TABLE src_result (
            `id` STRING,
            input_md5 STRING,
            output_md5 STRING,
            log STRING,
            metric STRING,
            create_time TIMESTAMP(6),      
            update_time TIMESTAMP(6),
            workflow_id_id STRING,
            error_details STRING,
            error_stage STRING,
            error_type STRING,
            PRIMARY KEY (`id`) NOT ENFORCED
        ) WITH (
            'connector' = 'mysql-cdc',
            'hostname' = 'x',
            'port' = '3306',
            'username' = 'x',
            'password' = 'x',
            'database-name' = 'x',
            'table-name' = 'result'
        );
    """)

    table_env.execute_sql("CREATE CATALOG iceberg WITH ("
                      "'type'='iceberg', "
                      "'catalog-type'='hive', "
                      "'uri'='thrift://x',"
                      "'warehouse'='tos://x',"
                      "'format-version'='2')")
    
    
    table_env.execute_sql("""INSERT INTO iceberg.results select * from src_result;""")

初始同步正常，但在MySQL中删除一行后，Iceberg中出现2条相同的行；执行更新操作后，Iceberg中出现3条行（2条原数据、1条更新后数据）。

已知环境信息：

• Flink版本：1.16.2
• Flink CDC版本：flink-sql-connector-mysql-cdc-2.2.1
• MySQL版本：5.7
• Iceberg表格式版本为v2，支持行级更新与删除

可能的原因分析 #

• 目标Iceberg表未定义主键：尽管源MySQL CDC表声明了主键，但Iceberg表未配置主键约束。Flink无法识别需要更新/删除的目标行，会将UPDATE/DELETE事件直接转为INSERT操作，导致旧数据留存、新数据重复插入。
• 使用普通INSERT而非Upsert模式写入：INSERT INTO是纯追加模式，不处理任何变更逻辑。针对CDC流的更新/删除事件，必须启用Upsert模式，这依赖目标表的主键定义，以及Iceberg连接器的Upsert支持配置。
• MySQL binlog格式不正确：如果MySQL的binlog_format不是ROW格式，Flink CDC无法捕获行级的更新/删除事件，会导致事件类型识别错误，进而引发数据重复。需要确认MySQL配置中binlog_format=ROW。
• Iceberg表未显式开启Upsert支持：即使表是v2版本，部分场景下需要显式配置'write.upsert.enabled' = 'true'，确保Flink可以生成正确的Iceberg更新/删除文件，而非直接追加数据。

内容的提问来源于stack exchange，提问作者Martin_5125