寻找两个DataFrame的唯一行与差异单元格，支持乱序对比并高亮差异

问题分析 #

你之前用pd.concat([df1, df2]).drop_duplicates(keep=False)无法得到预期结果，核心问题在于：

• 原索引会干扰重复行判断，即使内容完全一致的行，索引不同也会被当成不同行
• 该方法无法区分「仅在单个DataFrame存在的行」和「同一标识行的单元格差异」两类情况

以下是满足需求的通用代码，支持任意数量DataFrame对比，无需硬编码列名，同时可标记两类差异并高亮单元格。

通用对比代码 #

import pandas as pd
from IPython.display import display

def compare_dfs(*dfs, df_names=None, key_cols=None):
    # 处理DataFrame命名，默认用"DataFrame 1/2/..."
    if df_names is None:
        df_names = [f"DataFrame {i+1}" for i in range(len(dfs))]
    
    # 校验所有DataFrame列结构一致
    base_cols = dfs[0].columns
    for df in dfs[1:]:
        if not df.columns.equals(base_cols):
            raise ValueError("所有待对比的DataFrame必须拥有完全一致的列")
    
    # 预处理：重置索引，避免原索引干扰
    processed_dfs = [df.copy().reset_index(drop=True) for df in dfs]
    
    # --- 1. 识别仅在单个DataFrame存在的行 ---
    combined_list = []
    for name, df in zip(df_names, processed_dfs):
        temp_df = df.copy()
        temp_df['来源'] = name
        combined_list.append(temp_df)
    combined_df = pd.concat(combined_list, ignore_index=True)
    
    # 确定分组列：未指定主键则用所有列，指定则用主键列
    group_cols = base_cols.tolist() if key_cols is None else key_cols
    
    # 统计每个分组的来源数量
    source_counts = combined_df.groupby(group_cols)['来源'].nunique()
    
    # 提取仅存在于单个DataFrame的行
    unique_row_groups = source_counts[source_counts == 1].index.to_frame(index=False)
    only_in_df = combined_df.merge(unique_row_groups, on=group_cols, how='inner')
    only_in_df['差异类型'] = '仅存在于' + only_in_df['来源']
    
    # --- 2. 识别单元格内容不匹配的行（需指定主键列） ---
    cell_diff_df = pd.DataFrame()
    if key_cols is not None:
        # 提取在多个DataFrame中存在的主键分组
        common_key_groups = source_counts[source_counts >= 2].index.to_frame(index=False)
        common_rows = combined_df.merge(common_key_groups, on=group_cols, how='inner')
        
        # 遍历每个主键分组，对比单元格差异
        for _, group in common_rows.groupby(group_cols):
            # 按来源转宽表，便于对比同一主键的不同值
            wide_group = group.set_index(['来源'] + group_cols).unstack('来源')
            # 检查每一列是否存在多值差异
            for col in base_cols:
                if len(wide_group[col].unique()) > 1:
                    # 生成差异记录
                    diff_records = group.copy()
                    diff_records['差异类型'] = f"单元格差异: {col}"
                    diff_records[f"{col}_差异值"] = diff_records[col]
                    cell_diff_df = pd.concat([cell_diff_df, diff_records], ignore_index=True)
    
    # --- 3. 合并所有差异结果 ---
    df3 = pd.concat([only_in_df, cell_diff_df], ignore_index=True)
    if not df3.empty:
        # 调整列顺序，优先显示差异相关信息
        diff_related_cols = ['差异类型', '来源']
        value_cols = base_cols.tolist()
        extra_cols = [col for col in df3.columns if col.endswith('_差异值')]
        df3 = df3[diff_related_cols + value_cols + extra_cols]
    
    # --- 4. 高亮差异单元格（次要需求） ---
    def highlight_diff_cells(row):
        if '单元格差异' in row['差异类型']:
            target_col = row['差异类型'].split(': ')[1]
            return ['background-color: #fff3cd' if col == target_col else '' for col in row.index]
        return [''] * len(row)
    
    styled_df3 = df3.style.apply(highlight_diff_cells, axis=1)
    
    return df3, styled_df3

示例用法 #

示例1：双表对比（匹配你的需求示例）

# 构造示例DataFrame A
df_a = pd.DataFrame({
    'name': ['Ram', 'Sham', 'Buv', 'Raj', 'Tom'],
    'address': ['Delhi', 'Mumbai', 'Chennai', 'Kolkata', 'Bangalore'],
    'age': [25, 30, 35, 40, 45]
})

# 构造示例DataFrame B
df_b = pd.DataFrame({
    'name': ['Sham', 'Buv', 'Raj'],
    'address': ['Mumbai', 'Hyderabad', 'Kolkata'],
    'age': [30, 35, 40]
})

# 执行对比，指定name作为主键列（用于识别同一行）
diff_result, styled_diff = compare_dfs(df_a, df_b, df_names=['DataFrame A', 'DataFrame B'], key_cols=['name'])

# 查看结果
print("差异结果DataFrame:")
display(diff_result)
print("高亮差异单元格后的结果:")
display(styled_diff)

示例2：多表对比（支持任意数量DataFrame）

# 构造第三个DataFrame
df_c = pd.DataFrame({
    'name': ['Sham', 'Buv', 'Raj', 'Bob'],
    'address': ['Mumbai', 'Chennai', 'Kolkata', 'London'],
    'age': [30, 35, 41, 50]
})

# 三表对比
diff_result, styled_diff = compare_dfs(df_a, df_b, df_c, df_names=['A', 'B', 'C'], key_cols=['name'])
display(diff_result)

关键说明 #

1. 索引问题处理：所有DataFrame都会先重置索引，彻底避免原索引对重复行判断的干扰
2. 两类差异区分：
   • 「仅存在于单个DataFrame的行」：通过分组统计来源数量识别，未指定主键时，以整行内容为判断依据
   • 「单元格差异行」：需指定主键列（如示例中的name），用于识别同一主体的行，再对比其他列的内容差异
3. 高亮功能：通过Styler给差异单元格添加浅黄色背景，直观展示差异位置

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