如何封装可处理多薪资列的ETL函数以对比两日数据差异？

核心思路 #

通过ID列建立基准日薪资的映射字典，遍历当日数据逐一对比对应ID的薪资值，筛选出薪资变动的行索引，最终返回当日数据中变动行的DataFrame，自动适配两日数据行数量不一致的场景。

原单列处理代码参考 #

# 基准日数据与当日数据
base_df = pd.DataFrame(...)
current_df = pd.DataFrame(...)

# 构建基准日Payrate1的ID-薪资映射
base_pay_map = base_df.set_index('ID')['Payrate1'].to_dict()
changed_rows = []

# 遍历当日数据找变动行
for idx, row in current_df.iterrows():
    id_val = row['ID']
    current_pay = row['Payrate1']
    base_pay = base_pay_map.get(id_val, None)
    if base_pay is not None and current_pay != base_pay:
        changed_rows.append(idx)

# 生成结果DataFrame
result_df = current_df.loc[changed_rows]

封装后的复用函数 #

import pandas as pd

def find_changed_pay_rows(base_df, current_df, id_col, pay_col):
    # 构建基准日薪资的ID映射字典
    base_pay_map = base_df.set_index(id_col)[pay_col].to_dict()
    changed_indices = []
    
    # 遍历当日数据，对比薪资变动
    for idx, row in current_df.iterrows():
        current_id = row[id_col]
        current_pay = row[pay_col]
        base_pay = base_pay_map.get(current_id)
        
        # 仅记录基准日存在且薪资不一致的行
        if base_pay is not None and current_pay != base_pay:
            changed_indices.append(idx)
    
    # 返回变动行的DataFrame副本，避免原数据修改
    return current_df.loc[changed_indices].copy()

函数使用示例 #

# 构造测试数据
base_df = pd.DataFrame({
    'ID': [101, 102, 103],
    'Payrate1': [5000, 6000, 7000],
    'Payrate2': [1500, 1800, 2100]
})

current_df = pd.DataFrame({
    'ID': [101, 102, 104],
    'Payrate1': [5200, 6000, 8000],
    'Payrate2': [1500, 1900, 2400]
})

# 对比Payrate1的变动
result_pay1 = find_changed_pay_rows(base_df, current_df, 'ID', 'Payrate1')
print("Payrate1变动结果：")
print(result_pay1)

# 对比Payrate2的变动
result_pay2 = find_changed_pay_rows(base_df, current_df, 'ID', 'Payrate2')
print("\nPayrate2变动结果：")
print(result_pay2)

扩展说明 #

• 若需要将新增ID（基准日不存在的ID）也视为变动，可修改判断条件为：

  if base_pay is None or current_pay != base_pay:
      changed_indices.append(idx)
  

• 函数返回的是当日数据的子集，保留原索引和所有列，方便后续处理。

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