大数据集下关联两表并获取小于A表日期的B表最新行优化方案

数据集 #

DF_A（每日生成数据） #

ID  name    qty   date
1   abc     20   17/01/2022
1   abc     10   18/01/2022
2   def     10   24/01/2022
2   def     40   25/01/2022
2   def     67   26/01/2022

DF_B（价格历史数据） #

ID  name    price_dt    price
1   abc   18/01/2022    23.56
1   abc   17/01/2022    10.56
1   abc   16/01/2022    44.33
1   abc   15/01/2022    56.11
2   def   25/01/2022    2.98
2   def   26/01/2022    4.92
2   def   27/01/2022    4.88
2   def   24/01/2022    3.33
2   def   23/01/2022    8.47
2   def   22/01/2022    3.89

需求 #

为DF_A的每一行，匹配DF_B中小于当前行date的最新price_dt记录，最终预期结果：

ID  name    qty    date     price_dt    price
1   abc    20   17/01/2022  16/01/2022  44.33
1   abc    10   18/01/2022  17/01/2022  10.56
2   def    10   24/01/2022  23/01/2022  8.47
2   def    40   25/01/2022  24/01/2022  3.33
2   def    67   26/01/2022  25/01/2022  2.98

原方案的问题 #

原代码通过全量关联后排序去重的方式，会生成大量中间数据（示例中就产生40行），当数据集规模较大时，内存占用会直接超出阈值，无法运行：

DF_C = pd.merge(DF_A, DF_B, on='ID', how='left')
Expected_DF = DF_C.sort_values(by=['price_dt'], ascending=False)
Expected_DF = Expected_DF.drop_duplicates(subset=['ID','name','date'], keep='first')

低内存优化方案 #

方法1：使用merge_asof（首推，内存友好） #

merge_asof是pandas专门为时间序列向前匹配设计的API，不需要生成全量笛卡尔积的中间表，直接按顺序为每个DF_A行匹配符合条件的最新DF_B记录，内存占用仅为两个数据集本身的大小，完全适配超大规模数据场景。

代码实现：

import pandas as pd

# 将日期列转换为datetime类型，确保日期比较逻辑正确
DF_A['date'] = pd.to_datetime(DF_A['date'], format='%d/%m/%Y')
DF_B['price_dt'] = pd.to_datetime(DF_B['price_dt'], format='%d/%m/%Y')

# 按分组键+日期排序（merge_asof要求两边数据集都按此规则排序）
DF_A_sorted = DF_A.sort_values(by=['ID', 'name', 'date'])
DF_B_sorted = DF_B.sort_values(by=['ID', 'name', 'price_dt'])

# 执行向前匹配，匹配同ID、name下price_dt <= date的最新记录
Expected_DF = pd.merge_asof(
    DF_A_sorted,
    DF_B_sorted,
    by=['ID', 'name'],  # 分组匹配的键
    left_on='date',     # 左表用于匹配的日期列
    right_on='price_dt',# 右表用于匹配的日期列
    direction='backward'# backward表示找小于等于左表日期的最大右表日期
)

# 过滤出严格小于date的记录（如果需求是严格小于的话）
Expected_DF = Expected_DF[Expected_DF['price_dt'] < Expected_DF['date']]

# 可选：将日期转换回原格式
Expected_DF['date'] = Expected_DF['date'].dt.strftime('%d/%m/%Y')
Expected_DF['price_dt'] = Expected_DF['price_dt'].dt.strftime('%d/%m/%Y')

print(Expected_DF)

方法2：预处理DF_B缩减规模后关联 #

如果不想使用merge_asof，可以先预处理DF_B，只保留每个ID+name分组下，可能被DF_A匹配到的有效记录，再进行关联，也能大幅降低内存占用。

代码实现：

import pandas as pd

# 转换日期格式
DF_A['date'] = pd.to_datetime(DF_A['date'], format='%d/%m/%Y')
DF_B['price_dt'] = pd.to_datetime(DF_B['price_dt'], format='%d/%m/%Y')

# 按ID+name分组，收集DF_A中所有的日期
df_a_group_dates = DF_A.groupby(['ID', 'name'])['date'].apply(list).reset_index(name='target_dates')

# 预处理DF_B：为每个ID+name分组，筛选出所有小于目标日期的最新price_dt记录
def filter_valid_prices(group):
    group_sorted = group.sort_values('price_dt')
    target_dates = group['target_dates'].iloc[0]
    valid_records = []
    for date in target_dates:
        # 筛选出小于当前目标日期的所有price_dt，取最新的一条
        mask = group_sorted['price_dt'] < date
        if mask.any():
            valid_records.append(group_sorted[mask].iloc[-1])
    return pd.DataFrame(valid_records)

# 合并分组日期到DF_B，然后分组过滤
df_b_with_targets = pd.merge(DF_B, df_a_group_dates, on=['ID', 'name'])
df_b_filtered = df_b_with_targets.groupby(['ID', 'name']).apply(filter_valid_prices).reset_index(drop=True)

# 最后用merge_asof得到结果（此处用merge_asof更高效）
DF_A_sorted = DF_A.sort_values(by=['ID', 'name', 'date'])
df_b_filtered_sorted = df_b_filtered.sort_values(by=['ID', 'name', 'price_dt'])

Expected_DF = pd.merge_asof(
    DF_A_sorted,
    df_b_filtered_sorted,
    by=['ID', 'name'],
    left_on='date',
    right_on='price_dt',
    direction='backward'
)

# 转换回原日期格式
Expected_DF['date'] = Expected_DF['date'].dt.strftime('%d/%m/%Y')
Expected_DF['price_dt'] = Expected_DF['price_dt'].dt.strftime('%d/%m/%Y')

print(Expected_DF)

方案对比 #

• 方法1的merge_asof是最优解，代码简洁，执行效率最高，内存占用最少，适合所有规模的数据集。
• 方法2适合无法使用merge_asof的场景（比如旧版本pandas），但步骤相对繁琐，效率略低于方法1。

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