Flink-CDC同步MySQL到Iceberg后的数据校验方案对比及优化咨询

问题背景 #

我有一个10TB的MySQL数据库，已通过Flink-CDC将全量数据同步为S3上的Iceberg表，需要校验同步后是否存在数据丢失或值不匹配的情况，目前有两种PySpark实现方案：

初始化代码 #

# initialize two table in spark
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, concat_ws, md5

spark = SparkSession.builder.getOrCreate()

table_name = 'target_table'
df_iceberg_table = read_iceberg_table_using_spark(table_name)
df_mysql_table = read_mysql_table_using_spark(table_name)
table_columns = get_table_columns(table_name)

方案一：行哈希值对比 #

df_mysql_table_hash = (
    df_mysql_table
        .select(
            col('id'),
            md5(concat_ws('|', *table_columns)).alias('hash')
        )
)

df_iceberg_table_hash = (
    df_iceberg_table
        .select(
            col('id'),
            md5(concat_ws('|', *table_columns)).alias('hash')
        )
)

df_mysql_table_hash.createOrReplaceTempView('mysql_table_hash')
df_iceberg_table_hash.createOrReplaceTempView('iceberg_table_hash')

df_diff = spark.sql('''
    select 
        d1.id as mysql_id, 
        d2.id as iceberg_id, 
        d1.hash as mysql_hash, 
        d2.hash as iceberg_hash
    from mysql_table_hash d1
    left outer join iceberg_table_hash d2 on d1.id = d2.id
    where false
        or d2.id is null
        or d1.hash <> d2.hash
''')

# save df_diff to some where

方案二：使用PySpark subtract函数 #

df_diff = df_mysql_table.subtract(df_iceberg_table)

# save df_diff to some where

两种方案对比 #

性能（速度） #

方案一远快于方案二，核心原因是数据传输量的差异：

• 方案一只需要处理id和hash两列，单条数据体积极小，shuffle和存储开销大幅降低，对于10TB级别的大表，这个优势会被无限放大；
• 方案二的subtract需要对比全量列的完整数据，相当于要传输和处理两份10TB级别的数据，shuffle开销巨大，极易出现资源耗尽或超时。

功能完整性 #

• 方案一：能明确区分两种异常：Iceberg缺失的行（d2.id is null）和数据值不匹配的行（hash不同），还可通过反向join检查Iceberg是否存在多余行；
• 方案二：只能找出「MySQL有但Iceberg没有」的行，无法区分是丢失还是值不匹配，也无法发现Iceberg中多出来的脏数据（比如重复同步的行）。

潜在问题 #

• 方案一：需注意concat_ws的分隔符冲突（若列值本身包含|，会导致hash计算错误），以及空值的统一处理（不同数据源对空值的处理差异可能导致hash不一致）；
• 方案二：当表中有重复行时，无法精准定位异常行，且全量数据处理的资源成本极高。

更优的实现方法 #

针对10TB级大表，推荐基于方案一做优化，结合以下技巧：

1. 优化哈希计算逻辑

   • 替换分隔符：用不会出现在业务数据中的特殊字符（比如chr(0)空字符）代替|，避免拼接冲突；
   • 统一空值处理：用coalesce(col, '__NULL__')将所有空值转换为统一标识，消除数据源间的空值处理差异；
   • 选择高效哈希函数：对碰撞概率要求不高时，用Spark内置的hash()代替md5，计算速度提升数倍；需严格校验则保留md5。

2. 分区并行化处理

   • 按id范围将表拆分为多个批次（比如分100个批次），控制每个批次的数据量在Spark高效处理的范围内；
   • 利用Spark分区修剪特性，确保两边查询按相同分区规则读取，减少跨节点shuffle。

3. 双向校验

   • 在原有左join基础上，增加Iceberg表左join MySQL表的逻辑，检查是否存在Iceberg有但MySQL没有的行，确保数据双向一致。

4. 先粗后细的校验流程

   • 第一步：快速校验总行数是否一致，若行数差异大直接判定同步异常；
   • 第二步：行数一致时做抽样校验（比如随机抽1%数据），快速验证同步逻辑正确性；
   • 第三步：全量哈希校验放在低峰期执行，避免影响业务。

5. 利用Iceberg特性优化

   • 读取Iceberg同步完成的快照，避免读取未完成的同步数据；
   • 利用Iceberg分区统计信息，快速对比各分区行数，缩小异常排查范围。

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