Flink CDC数据湖迁移后数据一致性验证指南

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, concat_ws, md5

# 假设 SparkSession 已初始化
spark = SparkSession.builder.getOrCreate()

# 示例函数，实际需根据您的环境实现
def read_iceberg_table_using_spark(table_name):
    # 实际读取Iceberg表的逻辑，例如：
    # return spark.read.format("iceberg").load(f"s3://your-bucket/{table_name}")
    pass

def read_mysql_table_using_spark(table_name):
    # 实际读取MySQL表的逻辑，例如：
    # return spark.read.format("jdbc").option("url", "...").option("dbtable", table_name).load()
    pass

def get_table_columns(table_name):
    # 实际获取表所有列名的逻辑
    # 注意：应排除自增ID、时间戳等可能在CDC过程中自动变化的列，或确保它们在哈希计算时被统一处理
    return ["col1", "col2", "col3"] # 示例列名

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)

# 计算MySQL表的行哈希
df_mysql_table_hash = (
    df_mysql_table
        .select(
            col('id'),
            md5(concat_ws('|', *table_columns)).alias('hash')
        )
)

# 计算Iceberg表的行哈希
df_iceberg_table_hash = (
    df_iceberg_table
        .select(
            col('id'),
            md5(concat_ws('|', *table_columns)).alias('hash')
        )
)

# 创建临时视图用于SQL查询
df_mysql_table_hash.createOrReplaceTempView('mysql_table_hash')
df_iceberg_table_hash.createOrReplaceTempView('iceberg_table_hash')

# 执行SQL查询找出差异
df_diff_hash_comparison = 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 
        d2.id IS NULL             -- 目标表缺失的行
        OR d1.hash <> d2.hash     -- 哈希值不匹配的行
''')

# 展示或保存差异数据
if df_diff_hash_comparison.count() > 0:
    print("通过哈希值对比发现数据差异：")
    df_diff_hash_comparison.show()
else:
    print("通过哈希值对比，源表与目标表数据一致。")

# df_diff_hash_comparison.write.format("iceberg").mode("append").save("s3://your-bucket/data_diffs")

# 假设 df_mysql_table 和 df_iceberg_table 已初始化
# df_mysql_table = read_mysql_table_using_spark(table_name)
# df_iceberg_table = read_iceberg_table_using_spark(table_name)

# 找出MySQL中有，但Iceberg中没有的行（潜在的数据丢失）
df_diff_mysql_only = df_mysql_table.subtract(df_iceberg_table)

if df_diff_mysql_only.count() > 0:
    print("在MySQL中存在但在Iceberg中缺失的行：")
    df_diff_mysql_only.show()
else:
    print("Iceberg中不存在MySQL中独有的行。")

# 找出Iceberg中有，但MySQL中没有的行（潜在的脏数据或额外数据）
# 注意：这需要反向操作
df_diff_iceberg_only = df_iceberg_table.subtract(df_mysql_table)

if df_diff_iceberg_only.count() > 0:
    print("在Iceberg中存在但在MySQL中缺失的行（可能为Iceberg独有）：")
    df_diff_iceberg_only.show()
else:
    print("MySQL中不存在Iceberg中独有的行。")

# 假设 df_mysql_table 和 df_iceberg_table 已初始化

# 找出MySQL中有，但Iceberg中没有的行（包括重复行的差异）
diff_mysql_except_iceberg = df_mysql_table.exceptAll(df_iceberg_table)

if diff_mysql_except_iceberg.count() == 0:
    print("使用 exceptAll() 检查，MySQL中没有Iceberg中不存在的行。")
else:
    print("使用 exceptAll() 检查，MySQL中存在但在Iceberg中缺失的行（包括重复行差异）：")
    diff_mysql_except_iceberg.show()

# 找出Iceberg中有，但MySQL中没有的行（包括重复行的差异）
diff_iceberg_except_mysql = df_iceberg_table.exceptAll(df_mysql_table)

if diff_iceberg_except_mysql.count() == 0:
    print("使用 exceptAll() 检查，Iceberg中没有MySQL中不存在的行。")
else:
    print("使用 exceptAll() 检查，Iceberg中存在但在MySQL中缺失的行（包括重复行差异）：")
    diff_iceberg_except_mysql.show()

# 如果两个方向的 exceptAll() 结果都为空，则认为两个DataFrame完全相同
if diff_mysql_except_iceberg.count() == 0 and diff_iceberg_except_mysql.count() == 0:
    print("两个DataFrame在内容和重复行上完全一致。")