处理大文件时Python multiprocessing触发struct.error问题求助

嘿，这个问题我之前处理大数据集的时候也踩过坑！你遇到的struct.error本质是Python multiprocessing的一个隐藏限制：进程间通过管道传递数据时，会用struct.pack("!i")来打包消息的字节长度，而"i"是32位有符号整数，最大值是2147483647字节（差不多2GB）。哪怕你把DataFrame拆成1.5G的块，pickle序列化后的实际字节数可能还是超过了这个阈值，所以报错依然会出现。

下面给你几个实用的解决方案，按优先级排序：

1. 进一步拆分数据块，严格控制序列化后的体积 #

DataFrame经过pickle序列化后的大小通常比它在内存中的占用要大（尤其是包含字符串、分类类型的时候），所以不能只看原DataFrame的内存大小。你可以先测试单个拆分块序列化后的体积：

import pickle

# 拿其中一个拆分块测试序列化大小
test_chunk = np.array_split(data, 10)[0]
serialized_size = len(pickle.dumps(test_chunk))
print(f"单块序列化后大小：{serialized_size / (1024**3):.2f} GB")

根据测试结果，调整拆分的数量——比如如果单块序列化后是2.2GB，那你就拆成20块，确保每块序列化后远小于2GB。修改后的代码大概是这样：

# 拆分更多块，比如20块，确保单块序列化后<2GB
data_chunks = np.array_split(data, 20)
with mp.Pool(processes=5, maxtasksperchild=1) as pool1:
    pool1.map(write_in_parallel, data_chunks)
    pool1.close()
    pool1.join()

这个方案最简单，不需要改太多逻辑，适合快速解决问题。

2. 用共享内存传递数据，避免完整序列化 #

如果拆分太细影响处理效率，可以用共享内存让子进程直接读取数据，不用把整个DataFrame块序列化传递。比如用pandas自带的SharedMemory：

from multiprocessing import shared_memory
import pandas as pd
import numpy as np

def write_in_parallel(shm_name, df_shape, df_dtype, start_idx, end_idx):
    # 连接到共享内存
    shm = shared_memory.SharedMemory(name=shm_name)
    # 从共享内存中重建DataFrame
    df = pd.DataFrame(
        np.ndarray(df_shape, dtype=df_dtype, buffer=shm.buf),
        index=data.index,
        columns=data.columns
    )
    # 处理指定区间的数据
    chunk = df.iloc[start_idx:end_idx]
    # 这里写你的文件写入逻辑
    # chunk.to_csv(...)
    shm.close()

# 准备共享内存
shm = shared_memory.SharedMemory(create=True, size=data.nbytes)
# 将DataFrame的数据写入共享内存
df_shared = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
df_shared[:] = data.values[:]

# 拆分索引区间，而不是拆分DataFrame本身
total_rows = len(data)
chunk_count = 10
chunk_size = total_rows // chunk_count
task_args = [
    (shm.name, data.shape, data.dtype, i*chunk_size, (i+1)*chunk_size)
    for i in range(chunk_count)
]
# 最后一块处理剩余的行
task_args[-1] = (shm.name, data.shape, data.dtype, (chunk_count-1)*chunk_size, total_rows)

with mp.Pool(processes=5) as pool1:
    pool1.starmap(write_in_parallel, task_args)
    pool1.close()
    pool1.join()

# 最后一定要释放共享内存
shm.close()
shm.unlink()

这个方案的效率最高，因为子进程直接读取内存中的数据，没有序列化/反序列化的开销。

3. 用Manager托管大对象（适合小量超大块，效率略低） #

如果不想折腾共享内存，可以用multiprocessing.Manager——它会启动一个专门的服务器进程来管理对象，绕过管道的大小限制，不过因为多了一层进程通信，效率会稍差一点：

from multiprocessing import Manager

with Manager() as manager:
    # 把拆分后的DataFrame块存入Manager的列表
    shared_chunks = manager.list(np.array_split(data, 10))
    with mp.Pool(processes=5, maxtasksperchild=1) as pool1:
        pool1.map(write_in_parallel, shared_chunks)
        pool1.close()
        pool1.join()

最后还有个小提示：检查你的write_in_parallel函数，如果它不需要返回任何结果，最好让它返回None，避免子进程把处理后的大对象再传递回主进程，触发同样的大小限制。

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