如何快速检测Spark DataFrame差异？大表对比效率优化咨询

1. 先一次性筛选所有差异行（仅一次全表扫描） #

先通过全外关联找出两张表的所有差异行，避免多次重复扫描全表。如果表有唯一主键（如id），用主键关联效率最高；无主键时可通过整行哈希快速判断一致性：

# 假设id是表的唯一主键
diff_rows = spark.sql('''
    SELECT 
        t1.*, t2.*,
        CASE 
            WHEN t1.id IS NULL THEN 't2独有行' 
            WHEN t2.id IS NULL THEN 't1独有行' 
            ELSE '数据不一致行' 
        END AS diff_type
    FROM t1
    FULL OUTER JOIN t2 ON t1.id = t2.id
    WHERE t1.id IS NULL OR t2.id IS NULL OR t1 <> t2
''')

# 缓存差异行（差异行数量远小于全表，缓存后后续计算为内存级操作）
diff_rows.cache()
# 先统计总差异行数
total_diff = diff_rows.count()
print(f"总差异行数: {total_diff}")

2. 一次性统计所有列的差异次数 #

基于缓存的差异行，批量计算每列的差异计数，无需循环扫描全表：

from pyspark.sql.functions import col, sum, when

columns = t1.columns
# 生成每列的差异统计表达式（包含NULL值的判断）
diff_exprs = [
    sum(
        when(
            (col(f"t1.{c}") != col(f"t2.{c}")) | 
            (col(f"t1.{c}").isNull() != col(f"t2.{c}").isNull()), 
            1
        ).otherwise(0)
    ).alias(f"{c}_diff_count")
    for c in columns
]

# 计算差异汇总（含独有行统计+各列差异数）
diff_summary = diff_rows.agg(
    sum(when(col("diff_type") == "t1独有行", 1).otherwise(0)).alias("t1_unique_rows"),
    sum(when(col("diff_type") == "t2独有行", 1).otherwise(0)).alias("t2_unique_rows"),
    *diff_exprs
).collect()[0]

# 转换为字典输出结果
diff_result = dict(diff_summary.asDict())
print(diff_result)

核心优化逻辑 #

• 减少全表扫描次数：原方案循环30次，共触发60次全表扫描；新方案仅需1次关联扫描，后续计算基于小体量的差异行
• 缓存复用：差异行占比通常极低，缓存后所有后续计算均为内存级操作，速度大幅提升
• 处理NULL值：通过(a.isNull() != b.isNull())覆盖NULL值的差异场景，避免漏判

内容的提问来源于stack exchange，提问作者Golikov Andrey