如何在PySpark中对比两个DataFrame并定位不匹配的列？

方法1：逐列对比统计差异 #

遍历每个列，通过对比非空值数量、唯一值集合、聚合值等统计指标，快速定位不一致的列，适合初步排查问题。

示例代码：

def find_mismatched_columns(df1, df2):
    mismatched = []
    # 先校验Schema一致性
    assert set(df1.columns) == set(df2.columns), "两个DataFrame的Schema不一致"
    
    for col_name in df1.columns:
        # 对比非空值数量
        cnt1 = df1.filter(df1[col_name].isNotNull()).count()
        cnt2 = df2.filter(df2[col_name].isNotNull()).count()
        if cnt1 != cnt2:
            mismatched.append(f"列 {col_name}: 非空值数量不一致（df1: {cnt1}, df2: {cnt2}）")
            continue
        
        # 对比唯一值集合（小数据集适用，大数据集建议改用抽样）
        unique_vals1 = set(df1.select(col_name).distinct().rdd.flatMap(lambda x: x).collect())
        unique_vals2 = set(df2.select(col_name).distinct().rdd.flatMap(lambda x: x).collect())
        if unique_vals1 != unique_vals2:
            diff_vals = unique_vals1.symmetric_difference(unique_vals2)
            mismatched.append(f"列 {col_name}: 唯一值存在差异，差异值: {diff_vals}")
            continue
        
        # 数值列额外对比聚合值（求和/均值）
        col_type = df1.schema[col_name].dataType.typeName()
        if col_type in ["int", "long", "float", "double"]:
            sum1 = df1.agg({col_name: "sum"}).collect()[0][0]
            sum2 = df2.agg({col_name: "sum"}).collect()[0][0]
            if sum1 != sum2:
                mismatched.append(f"列 {col_name}: 求和值不一致（df1: {sum1}, df2: {sum2}）")
    
    return mismatched

# 使用示例
mismatch_list = find_mismatched_columns(df1, df2)
for item in mismatch_list:
    print(item)

方法2：关联后标记不匹配的行与列 #

如果需要定位到具体行和对应不匹配列，可以给DataFrame添加行标识后关联，逐列检查值是否相等，适合精准排查。

示例代码：

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

def find_mismatched_row_details(df1, df2):
    # 添加行ID（注意：仅当两个DataFrame分区逻辑一致时，该ID能对应到同一行）
    df1_with_id = df1.withColumn("row_id", monotonically_increasing_id())
    df2_with_id = df2.withColumn("row_id", monotonically_increasing_id())
    
    # 全外关联保留所有行
    joined_df = df1_with_id.join(df2_with_id, on="row_id", how="full_outer")
    
    # 生成列不匹配标记表达式
    mismatch_col_exprs = []
    for c in df1.columns:
        mismatch_flag = when(col(f"{c}_x") != col(f"{c}_y"), c).otherwise(None)
        mismatch_col_exprs.append(mismatch_flag.alias(f"{c}_mismatch"))
    
    # 筛选存在不匹配的行并整理结果
    mismatched_rows = joined_df.select("row_id", *mismatch_col_exprs).filter(
        col("row_id").isNotNull()
    ).collect()
    
    result = []
    for row in mismatched_rows:
        row_id = row["row_id"]
        cols_mismatch = [c for c in df1.columns if row[f"{c}_mismatch"] is not None]
        if cols_mismatch:
            result.append(f"行ID {row_id}: 不匹配列 {cols_mismatch}")
    
    return result

# 使用示例
mismatch_details = find_mismatched_row_details(df1, df2)
for item in mismatch_details:
    print(item)

注意事项 #

• 超大型数据集避免直接使用collect()，建议先通过sample(0.1)抽样后再对比，或仅保留关键统计指标的校验逻辑。
• 如果DataFrame有业务主键，优先用主键关联替代monotonically_increasing_id()，避免因分区差异导致的行匹配错误。

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