基于source列对比actual_data与pipeline_data列的Pandas实现方案

核心思路 #

针对不同source对应的actual_data和pipeline_data格式差异，通过自定义对比函数+按行应用的方式实现精准对比，最终生成result列。

步骤1：构造示例数据 #

先模拟符合场景的测试数据，方便后续演示：

import pandas as pd

data = {
    'source': ['A', 'B', 'A', 'C', 'B'],
    'actual_data': ['2024-05-20', 150, '2024-05-21', {'id': 1, 'name': 'test'}, 145],
    'pipeline_data': ['2024-05-20', 150, '2024-05-20', {'id': 1, 'name': 'test'}, 150]
}
df = pd.DataFrame(data)

步骤2：编写自定义对比函数 #

根据不同source的格式特点，编写针对性的对比逻辑：

def compare_data(row):
    source = row['source']
    actual = row['actual_data']
    pipeline = row['pipeline_data']
    
    # 针对Source A：日期字符串精准匹配
    if source == 'A':
        return '匹配' if actual == pipeline else '不匹配'
    
    # 针对Source B：数值对比（可设置误差范围）
    elif source == 'B':
        # 示例：允许±2的误差，不需要的话直接用 actual == pipeline
        return '匹配' if abs(actual - pipeline) <= 2 else '不匹配'
    
    # 针对Source C：嵌套字典/JSON结构对比
    elif source == 'C':
        # 简单结构直接用==，复杂嵌套可参考进阶方案
        return '匹配' if actual == pipeline else '不匹配'
    
    # 其他未知Source：默认直接对比原始值
    else:
        return '匹配' if actual == pipeline else '不匹配'

步骤3：应用函数生成Result列 #

使用apply方法按行调用自定义函数，生成结果列：

df['result'] = df.apply(compare_data, axis=1)

执行后查看结果：

source          actual_data         pipeline_data result
0      A           2024-05-20           2024-05-20    匹配
1      B                 150                 150    匹配
2      A           2024-05-21           2024-05-20  不匹配
3      C  {'id': 1, 'name': 'test'}  {'id': 1, 'name': 'test'}    匹配
4      B                 145                 150  不匹配

进阶优化：处理复杂嵌套结构 #

如果source对应的数据是多层嵌套的字典/JSON，可使用deepdiff库实现深度对比：

1. 先安装依赖：pip install deepdiff
2. 修改对比函数：

from deepdiff import DeepDiff

def compare_data(row):
    source = row['source']
    actual = row['actual_data']
    pipeline = row['pipeline_data']
    
    if source == 'A':
        # 额外处理：统一转为datetime对象对比（避免字符串格式不一致问题）
        try:
            actual_dt = pd.to_datetime(actual)
            pipeline_dt = pd.to_datetime(pipeline)
            return '匹配' if actual_dt == pipeline_dt else '不匹配'
        except ValueError:
            return '格式错误'
    elif source == 'B':
        return '匹配' if abs(actual - pipeline) <= 2 else '不匹配'
    elif source == 'C':
        diff = DeepDiff(actual, pipeline, ignore_order=True)
        return '匹配' if not diff else f'不匹配: {diff}'
    else:
        return '匹配' if actual == pipeline else '不匹配'

注意事项 #

• 若存在数据类型不一致的情况（如字符串日期vs datetime对象），需先统一类型再对比
• 可根据实际业务需求，扩展更多source对应的对比逻辑
• 大数据量场景下，apply效率较低，可考虑用np.select替代（适合规则明确的简单场景）

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