如何用PySpark实现类似Pandas.DataFrame.compare的DataFrame对比？

Pandas的compare方法会对比结构一致的DataFrame，仅展示存在差异的行与列，并分左右两侧显示两边的对应值。PySpark没有直接等价的API，但可以通过以下步骤手动实现相同逻辑：

核心实现步骤 #

1. 确保两个DataFrame的列名、数据类型完全匹配（和Pandas compare的要求一致）
2. 以唯一标识列（无标识时可添加自增ID）为键，对两个DataFrame做全外连接
3. 遍历所有字段，生成左右两侧的对比列，并标记字段是否存在差异
4. 过滤出至少有一个字段存在差异的行
5. 整理结果格式，对齐Pandas compare的输出风格

代码示例 #

假设我们有两个结构相同的PySpark DataFrame df1和df2，且包含唯一主键id：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, lit, concat_ws

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

# 构造示例数据
data1 = [(1, "Alice", 25, "New York"), (2, "Bob", 30, "London"), (3, "Charlie", 35, "Paris")]
data2 = [(1, "Alice", 26, "New York"), (2, "Bob", 30, "London"), (3, "Charlie", 35, "Berlin")]
schema = ["id", "name", "age", "city"]
df1 = spark.createDataFrame(data1, schema=schema)
df2 = spark.createDataFrame(data2, schema=schema)

# 1. 全外连接两个DataFrame
joined_df = df1.alias("left").join(df2.alias("right"), on="id", how="outer")

# 2. 生成对比列与差异标记
compare_cols = [col("left.id").alias("id")]
diff_flags = []

for field in schema:
    if field == "id":
        continue
    # 生成左右侧字段值列
    left_col = col(f"left.{field}").alias(f"{field}_left")
    right_col = col(f"right.{field}").alias(f"{field}_right")
    compare_cols.extend([left_col, right_col])
    # 标记当前字段是否存在差异
    diff_flag = when(col(f"left.{field}") != col(f"right.{field}"), lit(1)).otherwise(lit(0)).alias(f"{field}_diff")
    diff_flags.append(diff_flag)

# 拼接所有列生成对比中间表
comparison_df = joined_df.select(compare_cols + diff_flags)

# 3. 过滤出存在差异的行
diff_rows = comparison_df.filter(concat_ws("", diff_flags) != "0")

# 4. 移除差异标记列，得到最终对比结果
result_df = diff_rows.select([col for col in comparison_df.columns if not col.endswith("_diff")])

result_df.show()

输出效果 #

运行上述代码后，会输出仅包含差异行的对比结果：

+---+---------+---------+--------+--------+---------+---------+
| id|name_left|name_right|age_left|age_right|city_left|city_right|
+---+---------+---------+--------+--------+---------+---------+
|  1|    Alice|    Alice|      25|      26| New York| New York|
|  3|  Charlie|  Charlie|      35|      35|    Paris|    Berlin|
+---+---------+---------+--------+--------+---------+---------+

扩展优化 #

• 无唯一主键时，可添加自增ID：df1 = df1.withColumn("id", monotonically_increasing_id())，同理处理df2
• 封装为可复用函数：

def compare_spark_df(df1, df2, id_col="id"):
    joined_df = df1.alias("left").join(df2.alias("right"), on=id_col, how="outer")
    compare_cols = [col(f"left.{id_col}").alias(id_col)]
    diff_flags = []
    
    for field in df1.columns:
        if field == id_col:
            continue
        left_col = col(f"left.{field}").alias(f"{field}_left")
        right_col = col(f"right.{field}").alias(f"{field}_right")
        compare_cols.extend([left_col, right_col])
        diff_flag = when(col(f"left.{field}") != col(f"right.{field}"), lit(1)).otherwise(lit(0)).alias(f"{field}_diff")
        diff_flags.append(diff_flag)
    
    comparison_df = joined_df.select(compare_cols + diff_flags)
    diff_rows = comparison_df.filter(concat_ws("", diff_flags) != "0")
    return diff_rows.select([col for col in comparison_df.columns if not col.endswith("_diff")])

调用compare_spark_df(df1, df2)即可直接得到对比结果。

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