Flink-CDC数据湖数据一致性校验：PySpark实践指南

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

# 假设已配置好SparkSession
spark = SparkSession.builder.appName("DataConsistencyCheck").getOrCreate()

# 示例函数：读取Iceberg表和MySQL表
# 实际应用中需要替换为具体的读取逻辑
def read_iceberg_table_using_spark(table_name):
    # 例如：spark.read.format("iceberg").load(f"s3://your_bucket/{table_name}")
    print(f"Reading Iceberg table: {table_name}")
    # 模拟数据
    data = [(1, "Alice", 25, "New York"), (2, "Bob", 30, "London"), (3, "Charlie", 35, "Paris")]
    columns = ["id", "name", "age", "city"]
    return spark.createDataFrame(data, columns)

def read_mysql_table_using_spark(table_name):
    # 例如：spark.read.format("jdbc").option(...).load()
    print(f"Reading MySQL table: {table_name}")
    # 模拟数据，包含一个不一致的行和一个缺失的行
    data = [(1, "Alice", 25, "New York"), (2, "Robert", 30, "London"), (4, "David", 40, "Berlin")]
    columns = ["id", "name", "age", "city"]
    return spark.createDataFrame(data, columns)

table_name = 'your_table'
df_iceberg_table = read_iceberg_table_using_spark(table_name)
df_mysql_table = read_mysql_table_using_spark(table_name)

# 获取表的所有列名（不包括主键或其他不需要参与哈希计算的列）
# 实际应用中需要根据表的schema动态获取
table_columns = [col_name for col_name in df_mysql_table.columns if col_name != 'id']

print("MySQL Table Data:")
df_mysql_table.show()
print("Iceberg Table Data:")
df_iceberg_table.show()

print("\n--- Method 1: Row Hashing Comparison ---")

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

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

# 使用SQL进行左外连接和比较
df_diff_hash = spark.sql(f'''
    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           -- Iceberg中缺失的行 (数据丢失)
        OR d1.hash <> d2.hash   -- 哈希值不匹配的行 (数据不一致)
''')

print("Differences found using Row Hashing:")
df_diff_hash.show()

# 示例：保存差异数据
# df_diff_hash.write.mode("overwrite").format("parquet").save("path/to/diff_hash_results")

print("\n--- Method 2: Using DataFrame.subtract() ---")

# 找出MySQL中有但Iceberg中没有的行（潜在的数据丢失或Iceberg中缺少的新增数据）
df_mysql_only = df_mysql_table.subtract(df_iceberg_table)
print("Rows in MySQL but not in Iceberg (potential loss or new data):")
df_mysql_only.show()

# 找出Iceberg中有但MySQL中没有的行（潜在的Iceberg中多余的数据或MySQL中已删除的数据）
df_iceberg_only = df_iceberg_table.subtract(df_mysql_table)
print("Rows in Iceberg but not in MySQL (potential extra data or deleted data):")
df_iceberg_only.show()

# 组合两种差异以获得全面的不一致视图
# df_diff_subtract = df_mysql_only.unionAll(df_iceberg_only)
# print("Combined differences using subtract():")
# df_diff_subtract.show()

# 示例：保存差异数据
# df_mysql_only.write.mode("overwrite").format("parquet").save("path/to/mysql_only_results")
# df_iceberg_only.write.mode("overwrite").format("parquet").save("path/to/iceberg_only_results")

print("\n--- Method 3: Using DataFrame.exceptAll() ---")

# 找出MySQL中有但Iceberg中没有的行（包括重复行）
diff_mysql_except = df_mysql_table.exceptAll(df_iceberg_table)
print("Rows in MySQL but not in Iceberg (using exceptAll):")
diff_mysql_except.show()

# 找出Iceberg中有但MySQL中没有的行（包括重复行）
diff_iceberg_except = df_iceberg_table.exceptAll(df_mysql_table)
print("Rows in Iceberg but not in MySQL (using exceptAll):")
diff_iceberg_except.show()

# 检查是否存在差异
if diff_mysql_except.count() == 0 and diff_iceberg_except.count() == 0:
    print("DataFrames are identical (including duplicates and order for practical purposes).")
else:
    print("DataFrames have differences.")
    print("MySQL only rows (from exceptAll):")
    diff_mysql_except.show()
    print("Iceberg only rows (from exceptAll):")
    diff_iceberg_except.show()

# 示例：保存差异数据
# diff_mysql_except.write.mode("overwrite").format("parquet").save("path/to/mysql_except_results")
# diff_iceberg_except.write.mode("overwrite").format("parquet").save("path/to/iceberg_except_results")