Python：合并带时间条件的大DataFrame并规避内存错误

核心问题分析 #

直接全量merge会产生大量笛卡尔积数据，触发内存错误；简单分块处理未优化中间数据，导致耗时过长且生成超大文件。解决方案需聚焦减少中间数据量、分而治之的分组处理，同时优化内存使用。

步骤1：预处理实验室数据（lab_df） #

先对lab_df做清洗和格式化，减少后续计算负担：

import pandas as pd
import numpy as np

# 定义血液参数列
lab_cols = [col for col in lab_df.columns if col not in ['ID_1', 'ID_2', 'Timestamp_Lab']]

# 1. 按ID分组+时间排序，保证时间顺序正确
lab_df = lab_df.sort_values(by=['ID_1', 'ID_2', 'Timestamp_Lab'])

# 2. 对每个血液参数列执行向前填充（上移填充缺失值）
lab_df[lab_cols] = lab_df.groupby(['ID_1', 'ID_2'])[lab_cols].ffill()

# 3. 删除所有血液参数都为空的无效行
lab_df = lab_df.dropna(subset=lab_cols, how='all')

# 4. 给每组内的有效记录按时间顺序编号（用于后续宽表命名）
lab_df['seq'] = lab_df.groupby(['ID_1', 'ID_2']).cumcount() + 1

步骤2：分组合并并转宽表 #

按(ID_1, ID_2)分组处理，避免全量数据加载：

# 定义单分组处理函数
def process_single_group(id1, id2):
    # 获取当前分组的事件数据
    event_subset = event_df[(event_df['ID_1'] == id1) & (event_df['ID_2'] == id2)].copy()
    if event_subset.empty:
        return event_subset
    
    # 获取当前分组的预处理后实验室数据
    lab_subset = lab_df[(lab_df['ID_1'] == id1) & (lab_df['ID_2'] == id2)].copy()
    if lab_subset.empty:
        return event_subset
    
    # 合并并过滤时间条件（仅保留采血时间早于事件时间的记录）
    merged = event_subset.merge(lab_subset, on=['ID_1', 'ID_2'], how='left')
    merged = merged[merged['Timestamp_Lab'] < merged['Timestamp']]
    
    if merged.empty:
        return event_subset
    
    # 转宽表：将同参数的多条记录转为带序号的列（如Hemoglobin_1、Hemoglobin_2）
    wide_data = merged.pivot_table(
        index=['ID_1', 'ID_2', 'Timestamp', 'Event'],
        columns='seq',
        values=lab_cols,
        aggfunc='first'
    ).reset_index()
    
    # 重命名列，规范格式
    wide_data.columns = [f'{col[0]}_{col[1]}' if col[1] != '' else col[0] for col in wide_data.columns]
    return wide_data

# 获取所有存在实验室数据的ID组合
event_id_pairs = event_df[['ID_1', 'ID_2']].drop_duplicates()
lab_id_pairs = lab_df[['ID_1', 'ID_2']].drop_duplicates()
common_pairs = pd.merge(event_id_pairs, lab_id_pairs, on=['ID_1', 'ID_2'])

# 处理所有有实验室数据的分组
processed_dfs = []
for _, (id1, id2) in common_pairs.iterrows():
    processed = process_single_group(id1, id2)
    processed_dfs.append(processed)

# 处理无实验室数据的分组（直接保留原事件数据）
no_lab_pairs = event_id_pairs[~event_id_pairs.set_index(['ID_1', 'ID_2']).index.isin(common_pairs.set_index(['ID_1', 'ID_2']).index)]
for _, (id1, id2) in no_lab_pairs.iterrows():
    processed_dfs.append(event_df[(event_df['ID_1'] == id1) & (event_df['ID_2'] == id2)].copy())

# 合并所有结果并清理全空列
final_result = pd.concat(processed_dfs, ignore_index=True)
final_result = final_result.dropna(axis=1, how='all')

# 可选：过滤所有血液参数都为空的行
final_result = final_result.dropna(subset=[col for col in final_result.columns if col.endswith(tuple(map(str, range(1, 100))))], how='all')

步骤3：内存优化（可选但关键） #

针对大数据量，进一步压缩数据类型减少内存占用：

# 压缩ID列类型（如果ID范围在int32范围内）
event_df[['ID_1', 'ID_2']] = event_df[['ID_1', 'ID_2']].astype('int32')
lab_df[['ID_1', 'ID_2']] = lab_df[['ID_1', 'ID_2']].astype('int32')

# 压缩血液参数列类型（如果精度允许，从float64转float32）
lab_df[lab_cols] = lab_df[lab_cols].astype('float32')

超大数据量进阶方案：使用Dask #

若数据集超出单机内存，用Dask自动分块并行处理，逻辑与Pandas一致但无需手动分块：

import dask.dataframe as dd

# 读取数据（替换为你的文件路径）
event_dd = dd.read_csv(
    'event_data.csv',
    parse_dates=['Timestamp'],
    dtype={'ID_1': 'int32', 'ID_2': 'int32', 'Event': 'int32'}
)
lab_dd = dd.read_csv(
    'lab_data.csv',
    parse_dates=['Timestamp_Lab'],
    dtype={'ID_1': 'int32', 'ID_2': 'int32'}
)

# 预处理lab_dd（逻辑与Pandas一致）
lab_cols = [col for col in lab_dd.columns if col not in ['ID_1', 'ID_2', 'Timestamp_Lab']]
lab_dd = lab_dd.sort_values(by=['ID_1', 'ID_2', 'Timestamp_Lab'])
lab_dd[lab_cols] = lab_dd.groupby(['ID_1', 'ID_2'])[lab_cols].ffill()
lab_dd = lab_dd.dropna(subset=lab_cols, how='all')
lab_dd['seq'] = lab_dd.groupby(['ID_1', 'ID_2']).cumcount() + 1

# 合并+过滤时间条件
merged_dd = event_dd.merge(lab_dd, on=['ID_1', 'ID_2'], how='left')
merged_dd = merged_dd[merged_dd['Timestamp_Lab'] < merged_dd['Timestamp']]

# 转宽表+重命名列
wide_dd = merged_dd.pivot_table(
    index=['ID_1', 'ID_2', 'Timestamp', 'Event'],
    columns='seq',
    values=lab_cols,
    aggfunc='first'
).reset_index()

def rename_cols(cols):
    return [f'{col[0]}_{col[1]}' if col[1] != '' else col[0] for col in cols]
wide_dd.columns = rename_cols(wide_dd.columns)

# 清理全空列并保存结果
wide_dd = wide_dd.dropna(axis=1, how='all')
wide_dd.to_csv('final_result_*.csv', index=False)

为什么这个方案更高效？ #

1. 分组拆分：避免全量笛卡尔积，每个分组仅处理小批量数据，内存压力骤降
2. 提前预处理：删除无效行、填充缺失值，减少后续计算的数据量
3. 按需生成宽表：仅对有匹配数据的分组转宽表，避免无意义的空列生成
4. Dask并行：超大数据量时自动分块并行，充分利用CPU资源

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