合并Pandas DataFrame时遇内存错误，求大数据处理方案

本地处理优化方案 #

方案1：时间索引+范围查询替代交叉连接 #

交叉连接会生成len(dfA)*len(dfB)行，大数据量下必然触发内存溢出，换用基于时间索引的范围匹配，只保留符合±10秒条件的行，内存占用大幅降低：

import pandas as pd

# 转换时间格式并排序（索引查询依赖有序性）
dfA['InsertionTime'] = pd.to_datetime(dfA['InsertionTime'])
dfB['InsertionTime'] = pd.to_datetime(dfB['InsertionTime'])
dfA = dfA.sort_values('InsertionTime').reset_index(drop=True)
dfB = dfB.sort_values('InsertionTime').reset_index(drop=True)

# 给dfB建立时间索引，加速范围查询
dfB = dfB.set_index('InsertionTime')
tolerance = pd.Timedelta(seconds=10)

result_list = []
# 遍历dfA的每一行，匹配dfB中符合时间范围的记录
for idx, row in dfA.iterrows():
    start_time = row['InsertionTime'] - tolerance
    end_time = row['InsertionTime'] + tolerance
    # 提取时间范围内的dfB数据
    matched = dfB.loc[start_time:end_time].reset_index()
    if not matched.empty:
        # 合并当前dfA行与匹配到的dfB行
        merged_row = pd.concat([pd.DataFrame([row]*len(matched)), matched], axis=1)
        result_list.append(merged_row)

# 合并所有结果
consolidated = pd.concat(result_list, ignore_index=True)

# 重命名列（与原逻辑一致）
consolidated = consolidated.rename(columns={
    'InsertionTime_x': 'InsertionTime_df1',
    'InsertionTime_y': 'InsertionTime_df2',
    'DeviceName_x': 'DeviceName_df1',
    'DeviceName_y': 'DeviceName_df2',
    'DeviceIP_x': 'DeviceIP_df1',
    'DeviceIP_y': 'DeviceIP_df2',
    'AlarmName_x': 'AlarmName_df1',
    'AlarmName_y': 'AlarmName_df2'
})

方案2：分块处理dfA #

如果dfA数据量极大，单遍遍历仍占内存，可将dfA拆分为小块逐个处理：

import pandas as pd

# 预处理步骤（同方案1）
dfA['InsertionTime'] = pd.to_datetime(dfA['InsertionTime'])
dfB['InsertionTime'] = pd.to_datetime(dfB['InsertionTime'])
dfA = dfA.sort_values('InsertionTime').reset_index(drop=True)
dfB = dfB.sort_values('InsertionTime').reset_index(drop=True)
dfB = dfB.set_index('InsertionTime')

tolerance = pd.Timedelta(seconds=10)
chunk_size = 10000  # 根据本地内存大小调整块尺寸
result_list = []

# 分块遍历dfA
for i in range(0, len(dfA), chunk_size):
    chunk = dfA.iloc[i:i+chunk_size]
    for idx, row in chunk.iterrows():
        start_time = row['InsertionTime'] - tolerance
        end_time = row['InsertionTime'] + tolerance
        matched = dfB.loc[start_time:end_time].reset_index()
        if not matched.empty:
            merged_row = pd.concat([pd.DataFrame([row]*len(matched)), matched], axis=1)
            result_list.append(merged_row)

consolidated = pd.concat(result_list, ignore_index=True)
# 重命名列（同方案1）

方案3：merge_asof高效匹配（适合一对一/一对多最近匹配） #

如果业务允许只匹配±10秒内最近的一条记录，merge_asof是性能最优的选择，无需生成交叉表：

import pandas as pd

# 必须先按时间排序
dfA['InsertionTime'] = pd.to_datetime(dfA['InsertionTime']).sort_values()
dfB['InsertionTime'] = pd.to_datetime(dfB['InsertionTime']).sort_values()

# 匹配±10秒内的最近记录
consolidated = pd.merge_asof(
    dfA, dfB,
    on='InsertionTime',
    tolerance=pd.Timedelta(seconds=10),
    direction='nearest'  # 可选：'backward'（取早于目标时间的最近值）/'forward'（取晚于的最近值）
)
# 重命名列（略）

数据库处理方案 #

如果本地内存无法支撑，将数据导入数据库用SQL查询是更稳妥的方案，以SQLite为例：

步骤1：导入数据到SQLite #

import pandas as pd
import sqlite3

# 转换时间列
dfA['InsertionTime'] = pd.to_datetime(dfA['InsertionTime'])
dfB['InsertionTime'] = pd.to_datetime(dfB['InsertionTime'])

# 连接本地SQLite数据库文件
conn = sqlite3.connect('time_match.db')
# 将DataFrame导入数据库表
dfA.to_sql('tableA', conn, if_exists='replace', index=False)
dfB.to_sql('tableB', conn, if_exists='replace', index=False)

步骤2：执行SQL查询获取匹配结果 #

query = """
SELECT 
    a.InsertionTime AS InsertionTime_df1,
    b.InsertionTime AS InsertionTime_df2,
    a.DeviceName AS DeviceName_df1,
    b.DeviceName AS DeviceName_df2,
    a.DeviceIP AS DeviceIP_df1,
    b.DeviceIP AS DeviceIP_df2,
    a.AlarmName AS AlarmName_df1,
    b.AlarmName AS AlarmName_df2
FROM tableA a
JOIN tableB b 
ON ABS(strftime('%s', a.InsertionTime) - strftime('%s', b.InsertionTime)) <= 10;
"""
# 导出查询结果到DataFrame
consolidated = pd.read_sql(query, conn)
conn.close()

若使用PostgreSQL等数据库，时间差语法可简化为：ON a.InsertionTime BETWEEN b.InsertionTime - INTERVAL '10 seconds' AND b.InsertionTime + INTERVAL '10 seconds'，性能更优，适合超大规模数据。

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