如何在PySpark中对比CSV生成的DataFrame并高亮列差异？

先确认结构一致性 #

首先得保证两个DataFrame的列名、数量完全匹配，否则没法直接对比。可以加个断言检查：

# 校验列集合是否一致
assert set(df1.columns) == set(df2.columns), "两个DataFrame列结构不匹配，无法对比"

关联两个DataFrame #

根据数据情况选择关联方式：

有主键的情况（推荐） #

如果数据有唯一主键（比如id列），直接用主键关联，确保对应行匹配：

# 以id为关联键，全外连接保留所有行
combined_df = df1.alias("df1").join(df2.alias("df2"), on="id", how="fullouter")

无主键的情况 #

如果没有主键，只能通过行号关联（前提是两个文件的行顺序完全一致）：

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

# 给两个DataFrame添加行号
df1_with_row = df1.withColumn("row_num", row_number().over(Window.orderBy(*df1.columns)))
df2_with_row = df2.withColumn("row_num", row_number().over(Window.orderBy(*df2.columns)))

# 用行号关联
combined_df = df1_with_row.alias("df1").join(df2_with_row.alias("df2"), on="row_num", how="fullouter")

生成差异标记列 #

遍历所有需要对比的列，生成专门的差异列，把两边不同的值拼在一起，相同则标记“一致”：

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

# 排除关联键（id或row_num），获取所有要对比的列
compare_cols = [col for col in df1.columns if col not in ["id", "row_num"]]

# 为每个列生成差异标记
for col_name in compare_cols:
    combined_df = combined_df.withColumn(
        f"{col_name}_diff",
        when(col(f"df1.{col_name}") != col(f"df2.{col_name}"),
             concat_ws(" vs ", col(f"df1.{col_name}"), col(f"df2.{col_name}"))
            ).otherwise("一致")
    )

筛选并展示差异结果 #

可以先加一个“整体状态”列判断该行是否有差异，再筛选出有差异的行展示：

from pyspark.sql.functions import expr

# 生成整体状态列：只要有一个列差异就标记为"存在差异"
final_df = combined_df.withColumn(
    "整体状态",
    expr(f"CASE WHEN {' OR '.join([f'{col}_diff != \"一致\"' for col in compare_cols])} THEN '存在差异' ELSE '无差异' END")
)

# 只展示有差异的行，truncate=False避免内容被截断
final_df.filter(col("整体状态") == "存在差异").show(truncate=False)

可选：终端高亮差异值 #

如果终端支持ANSI颜色，可以用UDF给差异值加上红色高亮，更醒目：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# 定义UDF给差异值加红色高亮
@udf(StringType())
def highlight_diff(value):
    if value != "一致":
        return f"\033[91m{value}\033[0m"  # ANSI红色代码
    return value

# 给所有差异列应用高亮
for col_name in compare_cols:
    final_df = final_df.withColumn(f"{col_name}_diff", highlight_diff(col(f"{col_name}_diff")))

# 再次展示有差异的行
final_df.filter(col("整体状态") == "存在差异").show(truncate=False)

内容的提问来源于stack exchange，提问作者Vikram Singh Yadav