如何在Python中基于布尔条件合并？大DataFrame合并避内存错误

嘿，这个问题我太熟了——处理超大DataFrame时，先全量合并再过滤绝对是触发MemoryError的经典操作！别担心，咱们有好几种办法能在合并过程中就把数据量压下来，避免内存爆掉。

方法1：分块处理，化整为零 #

核心思路是把其中一个DataFrame拆成小块，逐个和另一个DataFrame合并、过滤，最后把结果拼接起来。这样每次只处理一小部分数据，内存压力会小很多。

如果你的数据是从文件读取的，直接用pandas的分块读取功能最方便：

import pandas as pd

# 根据你的内存情况调整分块大小，比如1万行一块
chunk_size = 10_000
result_chunks = []

# 遍历df1的分块（如果df2也很大，可以反过来遍历df2）
for chunk in pd.read_csv('df1.csv', chunksize=chunk_size):
    # 合并当前分块与df2
    merged_chunk = pd.merge(chunk, df2, on=['id'])
    # 过滤时间差小于1小时的行（记得先确保datetime列是datetime类型）
    filtered_chunk = merged_chunk[
        (merged_chunk['datetime'] - merged_chunk['datetime2']).dt.total_seconds() < 3600
    ]
    # 将过滤后的小块存入结果列表
    result_chunks.append(filtered_chunk)

# 拼接所有小块得到最终结果
final_df = pd.concat(result_chunks, ignore_index=True)

方法2：用分布式计算库“偷懒”（Dask/PySpark） #

如果分块处理还是顶不住，试试分布式计算工具——比如Dask，它的API和pandas几乎一致，但会自动在后台分块处理数据，不会一次性把所有数据加载到内存。

import dask.dataframe as dd

# 用Dask读取数据（支持csv、parquet等多种格式）
ddf1 = dd.read_csv('df1.csv')
ddf2 = dd.read_csv('df2.csv')

# 合并并过滤，Dask会延迟执行，直到你调用compute()
merged_ddf = dd.merge(ddf1, ddf2, on=['id'])
filtered_ddf = merged_ddf[
    (merged_ddf['datetime'] - merged_ddf['datetime2']).dt.total_seconds() < 3600
]

# 计算得到最终结果（如果内存够可以转成pandas DataFrame，也可以直接输出到文件）
final_df = filtered_ddf.compute()

如果数据量达到TB级别，PySpark会是更合适的选择，不过它的学习曲线稍陡一点，但处理超大规模数据的能力拉满。

方法3：提前预处理，缩小合并基数 #

这个方法是我最推荐的——先通过时间分组减少需要合并的行数，从根源上降低内存占用。

核心逻辑：两个时间差小于1小时的行，它们的小时数要么相同，要么相差1。所以我们可以先按id和“小时”合并，再做精确过滤：

import pandas as pd

# 确保datetime列是datetime类型（如果还不是的话）
df1['datetime'] = pd.to_datetime(df1['datetime'])
df2['datetime2'] = pd.to_datetime(df2['datetime2'])

# 给两个DataFrame添加小时级别的分组列
df1['hour'] = df1['datetime'].dt.floor('H')
df2['hour'] = df2['datetime2'].dt.floor('H')

# 先按id和hour合并（这样合并的行数会比直接按id合并少很多）
merged_pre = pd.merge(df1, df2, on=['id', 'hour'])
# 再精确过滤时间差小于1小时的行
final_df = merged_pre[
    (merged_pre['datetime'] - merged_pre['datetime2']).dt.total_seconds() < 3600
]

# 删掉临时的hour列
final_df = final_df.drop('hour', axis=1)

如果怕漏掉跨小时的符合条件的行，还可以给df2的hour列扩展成当前小时±1：

# 给df2生成三个小时列：当前小时、前一小时、后一小时
df2_expanded = df2.assign(hour=df2['hour']).append(
    df2.assign(hour=df2['hour'] - pd.Timedelta(hours=1))
).append(
    df2.assign(hour=df2['hour'] + pd.Timedelta(hours=1))
).drop_duplicates()

# 再和df1按id、hour合并
merged_pre = pd.merge(df1, df2_expanded, on=['id', 'hour'])

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