如何用Pandas实现类SQL Upsert功能，基于最新数据更新行值？

针对月度销售文件存在12周回溯重叠、需保留最新文件数据的需求，以下是高效的Pandas实现方案：

核心思路 #

每个文件的文件名包含生成年月（如sales_mar2022.csv对应2022年3月），我们可以：

1. 为每行数据标记其来源文件的生成时间
2. 以Date和Product ID作为唯一标识分组
3. 保留每个分组中来源文件时间最新的行，确保数据是最准确的版本

具体实现步骤 #

1. 导入依赖并定义文件名解析函数 #

import pandas as pd
import glob
import re

def get_file_date(filename):
    # 从文件名提取年月信息并转为datetime格式
    match = re.search(r'sales_(\w{3}\d{4})\.csv', filename)
    if match:
        return pd.to_datetime(match.group(1), format='%b%Y')
    return pd.NaT

2. 批量读取文件并添加来源时间列 #

读取所有销售文件，同时为每行添加file_date列标记数据来源文件的生成时间，避免后续无法区分数据版本：

# 获取所有销售文件路径
file_paths = glob.glob('sales_*.csv')

dfs = []
for path in file_paths:
    # 读取文件时指定类型、解析日期，减少内存占用并避免类型错误
    df = pd.read_csv(
        path,
        dtype={'Product ID': str, 'Sales': float, 'Volume': int},
        parse_dates=['Date'],
        date_parser=lambda x: pd.to_datetime(x, format='%d/%m/%Y'),
        engine='pyarrow'  # 需安装pyarrow，提升读取效率
    )
    df['file_date'] = get_file_date(path)
    dfs.append(df)

# 合并所有DataFrame
combined_df = pd.concat(dfs, ignore_index=True)

3. 保留最新版本数据（两种高效方法） #

方法一：排序后去重（适合大数据量，操作直观）

通过按file_date降序排序，直接保留每组的第一行（最新数据）：

final_df = combined_df.sort_values(
    by=['Date', 'Product ID', 'file_date'],
    ascending=[True, True, False]
).drop_duplicates(subset=['Date', 'Product ID'], keep='first').drop(columns=['file_date'])

方法二：Groupby + idxmax（内存占用更低）

通过分组获取每个标识组中最新数据的索引，直接提取目标行：

# 确保file_date为datetime类型
combined_df['file_date'] = pd.to_datetime(combined_df['file_date'])

# 获取每个(Date, Product ID)组中最新数据的索引
latest_idx = combined_df.groupby(['Date', 'Product ID'])['file_date'].idxmax()

# 提取最终数据并清理列
final_df = combined_df.loc[latest_idx].reset_index(drop=True).drop(columns=['file_date'])

效率优化建议 #

• 指定数据类型：读取csv时明确各列类型，避免Pandas自动推断导致的内存浪费和类型错误
• 使用PyArrow引擎：读取csv时添加engine='pyarrow'，大幅提升读取速度并降低内存占用
• 分批处理：若文件数量极大导致内存不足，可分批读取并合并去重（每读N个文件就执行一次去重保留最新数据，再与下一批合并）
• 日期格式统一：确保Date列解析为datetime类型，避免字符串排序导致的错误

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