Flink CDC数据湖迁移后的数据一致性校验：PySpark实践与方法比较

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

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

# 模拟加载数据，实际中需根据具体连接器实现
def read_iceberg_table_using_spark(table_name):
    # 实际应通过Spark Catalog加载Iceberg表
    return spark.read.format("iceberg").load(f"s3://your_bucket/{table_name}")

def read_mysql_table_using_spark(table_name):
    # 实际应通过JDBC连接MySQL
    return spark.read.format("jdbc") \
        .option("url", "jdbc:mysql://your_mysql_host:3306/your_database") \
        .option("dbtable", table_name) \
        .option("user", "your_user") \
        .option("password", "your_password") \
        .load()

def get_table_columns(table_name):
    # 实际应从数据库或元数据服务获取列名
    # 这里假设我们知道需要校验的列
    return ['col1', 'col2', 'col3', 'id'] # 示例列，'id' 通常是主键

table_name = 'your_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) # 获取所有需要参与哈希计算的列

# 排除主键列，因为主键用于join，哈希值应基于其他数据列
data_columns_for_hash = [c for c in table_columns if c != 'id']

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

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

# 创建临时视图以便使用Spark SQL
df_mysql_table_hash.createOrReplaceTempView('mysql_table_hash')
df_iceberg_table_hash.createOrReplaceTempView('iceberg_table_hash')

# 找出差异行
df_diff_hash = spark.sql(f'''
    SELECT
        m.id AS mysql_id,
        i.id AS iceberg_id,
        m.hash AS mysql_hash,
        i.hash AS iceberg_hash
    FROM mysql_table_hash m
    LEFT OUTER JOIN iceberg_table_hash i ON m.id = i.id
    WHERE
        i.id IS NULL           -- 数据丢失：Iceberg中缺少该ID
        OR m.hash <> i.hash    -- 数据不匹配：哈希值不同
''')

# 显示差异或保存结果
if df_diff_hash.count() > 0:
    print("发现数据不一致或丢失：")
    df_diff_hash.show(truncate=False)
else:
    print("数据一致。")

# 也可以检查Iceberg中是否存在MySQL中没有的额外数据
df_extra_iceberg = spark.sql(f'''
    SELECT
        i.id AS iceberg_id,
        m.id AS mysql_id
    FROM iceberg_table_hash i
    LEFT OUTER JOIN mysql_table_hash m ON i.id = m.id
    WHERE
        m.id IS NULL           -- Iceberg中存在但MySQL中没有的额外数据
''')
if df_extra_iceberg.count() > 0:
    print("发现Iceberg中存在额外数据：")
    df_extra_iceberg.show(truncate=False)

# 假设 df_mysql_table 和 df_iceberg_table 已加载

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

# 找出Iceberg中有，但MySQL中没有的行（Iceberg中额外的数据）
df_diff_iceberg_only = df_iceberg_table.subtract(df_mysql_table)

if df_diff_mysql_only.count() > 0:
    print("发现MySQL中有但Iceberg中没有的行：")
    df_diff_mysql_only.show(truncate=False)
else:
    print("MySQL中的数据似乎都存在于Iceberg中。")

if df_diff_iceberg_only.count() > 0:
    print("发现Iceberg中有但MySQL中没有的额外行：")
    df_diff_iceberg_only.show(truncate=False)
else:
    print("Iceberg中没有MySQL中不存在的额外数据。")

# 假设 df_mysql_table 和 df_iceberg_table 已加载

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

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

if df_diff_mysql_only_all.count() > 0:
    print("发现MySQL中有但Iceberg中没有的行（包括重复）：")
    df_diff_mysql_only_all.show(truncate=False)
else:
    print("MySQL中的数据（包括重复）似乎都存在于Iceberg中。")

if df_diff_iceberg_only_all.count() > 0:
    print("发现Iceberg中有但MySQL中没有的额外行（包括重复）：")
    df_diff_iceberg_only_all.show(truncate=False)
else:
    print("Iceberg中没有MySQL中不存在的额外数据（包括重复）。")