如何使用pandas实现新旧数据表对比及自动更新函数

实现逻辑 #

• 第一步：将新旧表按主键ID进行外连接，为新旧表的同名字段添加后缀区分
• 第二步：基于外连接生成的_merge标识初步判断行状态：仅在新表存在标记为new、仅在旧表存在标记为inactive、两表都存在的行进入字段变更校验逻辑
• 第三步：对两表共有的行逐字段对比，任意字段值不一致则标记状态为edited，所有字段一致则标记为active
• 第四步：对标记为edited的行，修改的字段按OLD:旧值, NEW:新值格式拼接，未修改字段保留原值；其他状态行直接保留对应表的原始值
• 第五步：整理输出列，保留原始表的所有字段+新增的Status状态列

完整代码实现 #

import pandas as pd

def compare_dataframes(frame_old, frame_new, primary_key='ID'):
    # 获取所有非主键列
    compare_cols = [col for col in frame_old.columns if col != primary_key]
    # 外连接合并两表，区分新旧字段
    merged_df = frame_old.merge(
        frame_new,
        on=primary_key,
        how='outer',
        suffixes=('_old', '_new'),
        indicator=True
    )
    # 初始化状态列
    merged_df['Status'] = ''
    # 标记仅旧表存在的行
    merged_df.loc[merged_df['_merge'] == 'left_only', 'Status'] = 'inactive'
    # 标记仅新表存在的行
    merged_df.loc[merged_df['_merge'] == 'right_only', 'Status'] = 'new'
    # 处理两表都存在的行，判断是否有字段变更
    both_exist_mask = merged_df['_merge'] == 'both'
    
    def check_edited(row):
        for col in compare_cols:
            if str(row[f'{col}_old']) != str(row[f'{col}_new']):
                return True
        return False
    
    merged_df.loc[both_exist_mask, 'is_edited'] = merged_df[both_exist_mask].apply(check_edited, axis=1)
    merged_df.loc[both_exist_mask & (merged_df['is_edited'] == True), 'Status'] = 'edited'
    merged_df.loc[both_exist_mask & (merged_df['is_edited'] == False), 'Status'] = 'active'

    # 构建最终输出表
    result = merged_df[[primary_key, 'Status']].copy()
    for col in compare_cols:
        def get_col_value(row):
            if row['Status'] == 'inactive':
                return row[f'{col}_old']
            elif row['Status'] == 'new':
                return row[f'{col}_new']
            elif row['Status'] == 'active':
                return row[f'{col}_old']
            else: # edited状态
                if str(row[f'{col}_old']) == str(row[f'{col}_new']):
                    return row[f'{col}_old']
                else:
                    return f"OLD:{row[f'{col}_old']}, NEW:{row[f'{col}_new']}"
        result[col] = merged_df.apply(get_col_value, axis=1)
    
    return result.reset_index(drop=True)

# 测试用例
dict_old= {
           'ID' : ["IQL_000", "SID_002", "OLA_056"],
           'Text' : ["foo", "hello", "a descirption"],
           'Number' : ["0991", "1231", "333"]
          }
frame_old= pd.DataFrame(dict_old)

dict_new= {
           'ID' : ["IQL_000", "SID_002", "PPA_344"],
           'Text' : ["foo", "world", "nothing"],
           'Number' : ["0991", "1231", "978"]
          }
frame_new= pd.DataFrame(dict_new)

# 生成对比结果
frame_updated = compare_dataframes(frame_old, frame_new)
print(frame_updated)

输出验证 #

运行上述代码得到的frame_updated和需求的预期输出完全一致：

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