Flink CDC数据同步后的数据完整性校验：PySpark实践指南

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

# 初始化SparkSession
spark = SparkSession.builder.appName("DataValidation").getOrCreate()

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

def read_mysql_table_using_spark(table_name):
    # 示例：读取MySQL表
    # 注意：需要JDBC驱动，并配置好连接信息
    return spark.read.format("jdbc") \
        .option("url", "jdbc:mysql://localhost:3306/your_database") \
        .option("dbtable", table_name) \
        .option("user", "your_user") \
        .option("password", "your_password") \
        .load()

def get_table_columns(table_name):
    # 示例：获取表的所有列名，不包括主键'id'
    # 实际中可能需要从数据库元数据或DataFrame schema中获取
    if table_name == 'target_table':
        return ['col1', 'col2', 'col3'] # 假设的列名
    return []

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'), # 假设'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 = 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
    FULL OUTER JOIN iceberg_table_hash d2 ON d1.id = d2.id
    WHERE 
        d1.id IS NULL OR d2.id IS NULL OR d1.hash <> d2.hash
''')

# 显示差异结果
df_diff_hash.show()
# df_diff_hash.write.format(...).save(...) # 保存差异数据

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

# 找出Iceberg中有但在MySQL中没有的行（潜在的脏数据或额外数据）
df_extra_in_iceberg = df_iceberg_table.subtract(df_mysql_table)

print("MySQL中有但在Iceberg中没有的行 (数据丢失):")
df_missing_in_iceberg.show()

print("Iceberg中有但在MySQL中没有的行 (额外数据):")
df_extra_in_iceberg.show()

# 如果需要合并差异，可以对两者进行union
# df_diff_subtract = df_missing_in_iceberg.unionAll(df_extra_in_iceberg)
# df_diff_subtract.show()

# 找出df_mysql_table中有但在df_iceberg_table中没有的行，包括重复行
diff_mysql_to_iceberg = df_mysql_table.exceptAll(df_iceberg_table)

# 找出df_iceberg_table中有但在df_mysql_table中没有的行，包括重复行
diff_iceberg_to_mysql = df_iceberg_table.exceptAll(df_mysql_table)

print("MySQL中有但在Iceberg中没有的行 (包括重复行):")
diff_mysql_to_iceberg.show()

print("Iceberg中有但在MySQL中没有的行 (包括重复行):")
diff_iceberg_to_mysql.show()

# 检查是否存在差异
if diff_mysql_to_iceberg.count() == 0 and diff_iceberg_to_mysql.count() == 0:
    print("两个DataFrames完全相同 (包括行顺序和重复行)。")
else:
    print("两个DataFrames存在差异。")