Flink-CDC数据湖数据完整性校验：PySpark实践指南

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

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("DataValidation") \
    .config("spark.sql.catalog.iceberg", "org.apache.iceberg.spark.SparkSessionCatalog") \
    .config("spark.sql.catalog.iceberg.type", "hive") \
    .config("spark.sql.catalog.iceberg.uri", "thrift://localhost:9083") \
    .getOrCreate()

# 假设的函数，用于从Iceberg和MySQL读取数据
# 实际项目中需要根据具体连接器实现
def read_iceberg_table_using_spark(table_name):
    # 示例：读取Iceberg表
    return spark.read.format("iceberg").load(f"iceberg.{table_name}")

def read_mysql_table_using_spark(table_name):
    # 示例：读取MySQL表
    # 注意：对于10TB数据，直接全量读取MySQL可能效率低下，
    # 实际应考虑增量读取、快照读取或通过其他方式获取数据
    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(df):
    # 获取DataFrame的列名，排除主键或不参与哈希计算的列
    # 假设'id'是主键，且所有其他列都参与校验
    return [c for c in df.columns if c != '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(df_mysql_table) # 假设两表的列结构一致

# 为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')

# 找出差异行：
# 1. Iceberg中缺失的MySQL行 (d2.id is null)
# 2. 存在但哈希值不匹配的行 (d1.hash <> d2.hash)
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
    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.count() > 0:
    print("通过哈希值比较发现数据差异:")
    df_diff_hash.show(truncate=False)
else:
    print("通过哈希值比较，两表数据一致。")

# 可以将差异保存到文件系统或另一个表中
# df_diff_hash.write.mode("overwrite").format("parquet").save("path/to/diff_hash_results")

# 找出在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)

if df_missing_in_iceberg.count() > 0:
    print("在MySQL中存在但在Iceberg中缺失的行:")
    df_missing_in_iceberg.show(truncate=False)
else:
    print("Iceberg中没有缺失MySQL中的行。")

if df_extra_in_iceberg.count() > 0:
    print("在Iceberg中存在但在MySQL中缺失的行 (额外数据):")
    df_extra_in_iceberg.show(truncate=False)
else:
    print("Iceberg中没有额外的行。")

# 找出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)

if diff_mysql_to_iceberg.count() == 0 and diff_iceberg_to_mysql.count() == 0:
    print("使用 exceptAll() 比较，两表数据完全一致（包括重复行数量）。")
else:
    print("使用 exceptAll() 发现数据差异:")
    if diff_mysql_to_iceberg.count() > 0:
        print("\n在MySQL中存在但在Iceberg中缺失或数量不匹配的行:")
        diff_mysql_to_iceberg.show(truncate=False)
    if diff_iceberg_to_mysql.count() > 0:
        print("\n在Iceberg中存在但在MySQL中缺失或数量不匹配的行 (额外数据或数量不匹配):")
        diff_iceberg_to_mysql.show(truncate=False)