Python如何按唯一ID匹配对比两个CSV文件的逐列数据差异

核心实现思路 #

因为两个CSV的行顺序不固定，不能直接按行号逐行对比，核心要先靠唯一ID做行对齐，再逐列做值校验，整体逻辑走这几步即可：

• 先做前置校验：确认两个文件都存在指定ID列、ID无重复、两个文件的ID集合完全匹配，避免后续对比时出现找不到对应行的报错
• 把其中一个DataFrame设置ID为索引，避免每次匹配行都做全表扫描，数据量大的时候查询效率能提升几个量级
• 逐行遍历第一个文件，通过ID直接从索引后的第二个DataFrame取匹配行
• 逐列对比两行的字段值，单独做空值兼容（pandas的NaN、None、空字符串要统一判定为空，避免两边都是空值时误报差异），值不一致时抛出带ID、列名、两边实际值的断言信息，方便定位问题

现有代码可优化点 #

你贴的初版代码已经能实现基础对比逻辑，但是有几个可以调整的地方：

• 每次循环都用df2.loc[df2['id'] == Id]做全表布尔筛选，CSV行数多的时候运行速度会非常慢，提前建ID索引能大幅提速
• 对比逻辑写死了只对比column这一列，没法自动适配任意列结构的CSV，加个列遍历逻辑就能通用
• 断言报错只返回行索引，没有标注具体差异的ID、列名、两边的实际值，排查问题时需要手动回查，效率很低
• 没有处理ID在另一个文件中缺失的场景，遇到ID不匹配会直接抛索引错误，没有明确的错误提示
• 空值判断只覆盖了None、空字符串这类falsy值，没有兼容pandas读取CSV时生成的NaN值，会把两边都是NaN的情况误判为不一致

优化后可直接运行的代码 #

import pandas as pd

def compare_csv(file1_path: str, file2_path: str, id_col: str = 'id') -> None:
    # 读取两个CSV文件
    df1 = pd.read_csv(file1_path)
    df2 = pd.read_csv(file2_path)

    # 前置校验：ID列存在性检查
    if id_col not in df1.columns or id_col not in df2.columns:
        raise ValueError(f"两个CSV必须包含指定的ID列：{id_col}")
    
    # 前置校验：ID唯一性检查
    if not df1[id_col].is_unique:
        raise AssertionError(f"第一个文件的ID列[{id_col}]存在重复值，无法做唯一匹配")
    if not df2[id_col].is_unique:
        raise AssertionError(f"第二个文件的ID列[{id_col}]存在重复值，无法做唯一匹配")
    
    # 前置校验：ID集合一致性检查
    id1_set = set(df1[id_col])
    id2_set = set(df2[id_col])
    if id1_set != id2_set:
        err_info = []
        missing_in_f2 = id1_set - id2_set
        missing_in_f1 = id2_set - id1_set
        if missing_in_f2:
            err_info.append(f"以下ID在第二个文件中不存在：{sorted(missing_in_f2)}")
        if missing_in_f1:
            err_info.append(f"以下ID在第一个文件中不存在：{sorted(missing_in_f1)}")
        raise AssertionError("\n".join(err_info))
    
    # 将第二个文件设置ID为索引，提升行匹配效率
    df2_indexed = df2.set_index(id_col)
    # 提取需要对比的列（排除ID列本身）
    check_columns = [col for col in df1.columns if col != id_col]

    # 逐行逐列做断言校验
    for _, row in df1.iterrows():
        current_id = row[id_col]
        matched_row = df2_indexed.loc[current_id]

        for col in check_columns:
            val_f1 = row[col]
            val_f2 = matched_row[col]

            # 统一处理空值：两边都是空（NaN/None/去空格后空字符串）判定为一致
            val_f1_empty = pd.isna(val_f1) or str(val_f1).strip() == ""
            val_f2_empty = pd.isna(val_f2) or str(val_f2).strip() == ""
            if val_f1_empty and val_f2_empty:
                continue

            # 值不一致时抛出带详细定位信息的断言错误
            assert val_f1 == val_f2, (
                f"ID={current_id} 的行在列 [{col}] 存在差异：\n"
                f"文件1取值：{repr(val_f1)}\n"
                f"文件2取值：{repr(val_f2)}"
            )
    
    print("校验通过：两个CSV文件内容完全一致")

# 调用示例，替换成你的实际文件路径即可
# compare_csv("test1.csv", "test2.csv")

内容的提问来源于stack exchange，提问作者Raj Aryan Sharma