基于PySpark对比无主键同结构DataFrame的详细差异方法求助

需求说明 #

需要对比两个无主键的DataFrame，识别以下差异：

• 新增行（蓝色高亮）
• 删除行（黄色高亮）
• 单元格值变更（橙色高亮）
  同时需兼容：列无序、字符/数值混合类型、重复行、空值（null）场景。

实现方案 #

核心思路 #

由于无主键，通过行哈希值作为临时标识匹配行，统一列顺序后对比单元格差异，最后用Pandas样式系统实现高亮。

步骤与代码实现 #

1. 依赖导入

import pandas as pd
import hashlib

2. 预处理：对齐列与处理空值

先统一两个DataFrame的列顺序，将空值替换为统一标记（避免哈希计算时因null产生无意义差异）：

# 假设df_old为旧数据，df_new为新数据
common_cols = sorted(list(set(df_old.columns).union(df_new.columns)))

# 对齐列并填充空值
df_old_aligned = df_old.reindex(columns=common_cols).fillna("<NA>")
df_new_aligned = df_new.reindex(columns=common_cols).fillna("<NA>")

3. 生成行哈希（临时行标识）

用每行所有列的字符串拼接值生成哈希，作为临时匹配键：

# 矢量化生成哈希，提升大数据量处理效率
def generate_row_hash(df):
    str_rows = df.astype(str).agg("|".join, axis=1)
    return str_rows.apply(lambda x: hashlib.md5(x.encode()).hexdigest())

df_old_aligned["row_hash"] = generate_row_hash(df_old_aligned[common_cols])
df_new_aligned["row_hash"] = generate_row_hash(df_new_aligned[common_cols])

4. 标记差异类型

合并两个DataFrame，区分新增、删除、变更行：

# 标记数据来源
df_old_aligned["source"] = "old"
df_new_aligned["source"] = "new"

merged_df = pd.concat([df_old_aligned, df_new_aligned]).reset_index(drop=True)

# 统计哈希出现次数，区分匹配/不匹配行
hash_counts = merged_df["row_hash"].value_counts()
matched_hashes = hash_counts[hash_counts == 2].index  # 两端都存在的行（可能有变更）
unmatched_hashes = hash_counts[hash_counts == 1].index  # 仅一端存在的行（新增/删除）

# 分离差异行
deleted_rows = merged_df[(merged_df["source"] == "old") & (merged_df["row_hash"].isin(unmatched_hashes))]
added_rows = merged_df[(merged_df["source"] == "new") & (merged_df["row_hash"].isin(unmatched_hashes))]
changed_rows = merged_df[merged_df["row_hash"].isin(matched_hashes)]

5. 高亮样式定义

定义样式函数，对不同类型的差异应用对应颜色：

def highlight_differences(row):
    # 标记删除行（黄色）
    if row.name in deleted_rows.index:
        return ["background-color: #FFFF99"] * len(row)
    # 标记新增行（蓝色）
    elif row.name in added_rows.index:
        return ["background-color: #87CEEB"] * len(row)
    # 标记变更单元格（橙色）
    else:
        hash_val = row["row_hash"]
        if hash_val in matched_hashes:
            # 获取该行的新旧值对比
            old_row = changed_rows[(changed_rows["row_hash"] == hash_val) & (changed_rows["source"] == "old")][common_cols].iloc[0]
            new_row = changed_rows[(changed_rows["row_hash"] == hash_val) & (changed_rows["source"] == "new")][common_cols].iloc[0]
            styles = []
            for col in common_cols:
                if old_row[col] != new_row[col]:
                    styles.append("background-color: #FFA500")
                else:
                    styles.append("")
            # 忽略辅助列的样式
            styles += ["", ""]
            return styles
        return [""] * len(row)

6. 生成可视化差异报告

合并所有差异行，应用样式并导出/展示：

# 按删除→变更→新增的顺序整理差异
final_diff = pd.concat([deleted_rows, changed_rows, added_rows]).reset_index(drop=True)

# 应用样式并隐藏辅助列
styled_report = final_diff.style.apply(highlight_differences, axis=1)\
                                  .hide(["row_hash", "source"])

# 导出为HTML报告（方便查看）
styled_report.to_html("dataframe_diff_report.html")

# 直接在Jupyter中展示
display(styled_report)

优化说明 #

• 大数据量场景：改用pd.util.hash_pandas_object生成行哈希，速度更快：

  df_old_aligned["row_hash"] = pd.util.hash_pandas_object(df_old_aligned[common_cols], index=False)
  

• 重复行处理：完全相同的重复行将被视为匹配行，不会标记为差异；若需识别重复行数量变化，可额外统计哈希出现次数。

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