如何对比两个Python Pandas DataFrame并识别不匹配行与列

对比两个Pandas DataFrame，输出所有不匹配的行，并添加remark列：

• 若同一主键（id）的行存在列值差异，标注具体不匹配的列名
• 若某行仅存在其中一个DataFrame中，标注其归属（如only in source/only in target）

示例数据（更新版） #

source_df

id first_name last_name city salary
  1        AAA       FFF  bbb   1000
  2        BBB       GGG  sts   1000
  3        CCC       HHH  aaa   1000
  4        DDD       III  bbb   1000
  5        EEE       JJJ  sts   1000
  7        PPP       QQQ  aaa   5000
  8        lll       jjj        5000

target_df

id first_name last_name city salary
  1        AAA       FFF  bbb   2000
  2        BBB       GGG  sts   1000
  3        CCC       HHH  aaa   1000
  4        OOO       III  bbb   1000
  5        EEE       JJJ  tst   1000
  6        YYY       ZZZ  aaa   5000

期望输出 #

id first_name last_name city salary remark
 1        AAA       FFF  bbb   1000 salary
 1        AAA       FFF  bbb   2000 salary
 4        DDD       III  bbb   1000 first_name
 4        OOO       III  bbb   1000 first_name
 5        EEE       JJJ  sts   1000 city
 5        EEE       JJJ  tst   1000 city
 6        YYY       ZZZ  aaa   5000 only in target
 7        PPP       QQQ  aaa   5000 only in source
 8        lll       jjj        5000 only in source

以下是基于Pandas实现的代码，完全匹配需求：

import pandas as pd

def compare_dfs(source_df, target_df, key_col='id'):
    # 统一主键列类型，避免类型不匹配导致的匹配失败
    source_df[key_col] = source_df[key_col].astype(str)
    target_df[key_col] = target_df[key_col].astype(str)
    
    # 获取除主键外的所有对比列
    compare_cols = [col for col in source_df.columns if col != key_col]
    
    # 外连接合并两个DF，标记行的来源
    merged = pd.merge(
        source_df.assign(source=True),
        target_df.assign(target=True),
        on=key_col,
        how='outer',
        suffixes=('_source', '_target'),
        indicator=True
    )
    
    result_rows = []
    
    # 处理仅存在于单个DF的行
    for _, row in merged[merged['_merge'] != 'both'].iterrows():
        if row['_merge'] == 'left_only':
            # 提取source侧的行并添加标注
            source_row = row[[f"{col}_source" for col in source_df.columns]].rename(
                lambda x: x.replace('_source', '')
            )
            source_row['remark'] = 'only in source'
            result_rows.append(source_row)
        else:
            # 提取target侧的行并添加标注
            target_row = row[[f"{col}_target" for col in target_df.columns]].rename(
                lambda x: x.replace('_target', '')
            )
            target_row['remark'] = 'only in target'
            result_rows.append(target_row)
    
    # 处理两边都存在但有差异的行
    both_df = merged[merged['_merge'] == 'both']
    for _, row in both_df.iterrows():
        diff_cols = []
        # 逐列对比值，收集不匹配的列
        for col in compare_cols:
            source_val = row[f"{col}_source"]
            target_val = row[f"{col}_target"]
            # 跳过空值相等的情况
            if pd.isna(source_val) and pd.isna(target_val):
                continue
            if source_val != target_val:
                diff_cols.append(col)
        
        if diff_cols:
            # 添加source侧的差异行
            source_row = row[[f"{col}_source" for col in source_df.columns]].rename(
                lambda x: x.replace('_source', '')
            )
            source_row['remark'] = ', '.join(diff_cols)
            result_rows.append(source_row)
            # 添加target侧的差异行
            target_row = row[[f"{col}_target" for col in target_df.columns]].rename(
                lambda x: x.replace('_target', '')
            )
            target_row['remark'] = ', '.join(diff_cols)
            result_rows.append(target_row)
    
    # 整理结果：按主键排序，调整列顺序把remark放最后
    result_df = pd.DataFrame(result_rows).sort_values(by=key_col).reset_index(drop=True)
    cols = [col for col in result_df.columns if col != 'remark'] + ['remark']
    result_df = result_df[cols]
    
    return result_df

# ------------------------------
# 测试示例数据
# ------------------------------
source_df = pd.DataFrame({
    'id': [1,2,3,4,5,7,8],
    'first_name': ['AAA','BBB','CCC','DDD','EEE','PPP','lll'],
    'last_name': ['FFF','GGG','HHH','III','JJJ','QQQ','jjj'],
    'city': ['bbb','sts','aaa','bbb','sts','aaa',''],
    'salary': [1000,1000,1000,1000,1000,5000,5000]
})

target_df = pd.DataFrame({
    'id': [1,2,3,4,5,6],
    'first_name': ['AAA','BBB','CCC','OOO','EEE','YYY'],
    'last_name': ['FFF','GGG','HHH','III','JJJ','ZZZ'],
    'city': ['bbb','sts','aaa','bbb','tst','aaa'],
    'salary': [2000,1000,1000,1000,1000,5000]
})

# 执行对比并打印结果
result = compare_dfs(source_df, target_df)
print(result.to_string(index=False))

代码说明 #

1. 主键对齐：将主键列转换为字符串，避免因整数/字符串类型差异导致的匹配错误
2. 外连接合并：通过outer join整合两个DF，用_merge字段标记行的来源
3. 孤立行处理：直接提取仅存在于单个DF的行，添加对应归属标注
4. 差异行处理：逐列对比同主键的行，收集不匹配的列名，为两行分别添加标注
5. 结果整理：按主键排序，调整列顺序使remark列放在最后，提升可读性

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