Python Pandas多进程Pickle错误：无法序列化局部对象问题

我有一批存储股票交易数据的CSV文件，需要提取每个文件中'BABA'标的的数据段并合并为多日数据。单线程循环处理150个文件耗时约15分钟，尝试用multiprocessing加速时，触发错误：AttributeError: Can't pickle local object 'main.locals.compile'。

可运行的单线程代码 #

def main():

    import pandas as pd
    import glob
    import numpy as np
    import multiprocessing
    from multiprocessing import Pool

    path = '/Users/DataFiles' # 存储多个CSV文件的路径
    file_list = glob.glob(path + '/*.csv')
    stock_list = []

    def compile(file):
        df = pd.read_csv(file)
        df = df.loc[df['UnderlyingSymbol'] == 'BABA'] # 筛选标的为BABA的数据

#=== 对DataFrame做修改 ===

        def market_delta():
            return np.sum(df['Delta'] * df['OpenInterest']) / np.sum(df['OpenInterest'])
        df['DMarket'] = market_delta().round(4)

#=== 将处理后的DataFrame追加到列表 ===

        stock_list.append(df)

#=== 使用单线程循环处理 ===

    for file in file_list:
        compile(file)

#=== 合并DataFrame并导出 ===
stock_list_merged = pd.concat(stock_list)
stock_list_merged.to_csv('scholes expanded.csv', index = False)

if __name__ == '__main__':
    main()

报错的多进程代码 #

def main():

    import pandas as pd
    import glob
    import numpy as np
    import multiprocessing
    from multiprocessing import Pool

    path = '/Users/DataFiles' # 存储多个CSV文件的路径
    file_list = glob.glob(path + '/*.csv')
    stock_list = []

    def compile(file):
        df = pd.read_csv(file)
        df = df.loc[df['UnderlyingSymbol'] == 'BABA'] # 筛选标的为BABA的数据

#=== 对DataFrame做修改 ===

        def market_delta():
            return np.sum(df['Delta'] * df['OpenInterest']) / np.sum(df['OpenInterest'])
        df['DMarket'] = market_delta().round(4)

#=== 将处理后的DataFrame追加到列表 ===

        stock_list.append(df)

#=== 使用多进程处理 ===

    pool = Pool(processes = (multiprocessing.cpu_count()-1))
    results = pool.map(compile, file_list)
    pool.close()
    pool.join()
    results_df = pd.concat(results)

#=== 合并DataFrame并导出 ===
stock_list_merged = pd.concat(stock_list)
stock_list_merged.to_csv('scholes expanded.csv', index = False)

if __name__ == '__main__':
    main()

错误栈 #

Traceback (most recent call last):
File "/Users/andrewbochat/Desktop/OptionsData/practice multiprocessing.py", line 54, in 
main()
File "/Users/andrewbochat/Desktop/OptionsData/practice multiprocessing.py", line 41, in main
results = pool.map(compile, file_list)
File "/Users/andrewbochat/opt/anaconda3/lib/python3.9/multiprocessing/pool.py", line 364, in map
return self._map_async(func, iterable, mapstar, chunksize).get()
File "/Users/andrewbochat/opt/anaconda3/lib/python3.9/multiprocessing/pool.py", line 771, in get
raise self._value
File "/Users/andrewbochat/opt/anaconda3/lib/python3.9/multiprocessing/pool.py", line 537, in _handle_tasks
put(task)
File "/Users/andrewbochat/opt/anaconda3/lib/python3.9/multiprocessing/connection.py", line 211, in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "/Users/andrewbochat/opt/anaconda3/lib/python3.9/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
AttributeError: Can't pickle local object 'main.locals.compile'

错误原因 #

1. Pickle序列化限制：Python的multiprocessing依赖pickle在进程间传递函数和数据，但pickle无法序列化嵌套在函数内部的局部函数（比如原代码中main()里的compile函数），只能序列化模块级别的可调用对象。
2. 多进程内存隔离：原代码中试图让子进程往stock_list追加数据是无效的——每个子进程有独立的内存空间，主进程的stock_list不会被修改。
3. 逻辑冗余：代码同时使用了pool.map的返回值和stock_list，逻辑混乱。

修复方案 #

核心修改点 #

• 将compile函数移到模块级别（不要嵌套在main()内），让pickle能正常识别并序列化它。
• 让compile函数直接返回处理好的DataFrame，利用pool.map自动收集所有子进程的返回结果。
• 移除无效的stock_list，直接基于pool.map的返回值合并数据。

修正后的完整代码 #

import pandas as pd
import glob
import numpy as np
import multiprocessing
from multiprocessing import Pool

# 把compile函数移到模块级别，解决pickle序列化问题
def compile(file):
    df = pd.read_csv(file)
    df = df.loc[df['UnderlyingSymbol'] == 'BABA']
    
    # 简化market_delta计算，也可以保留原嵌套函数写法（不影响序列化）
    market_delta = np.sum(df['Delta'] * df['OpenInterest']) / np.sum(df['OpenInterest'])
    df['DMarket'] = market_delta.round(4)
    
    # 返回处理后的DataFrame，由pool.map统一收集
    return df

def main():
    path = '/Users/DataFiles'
    file_list = glob.glob(path + '/*.csv')

    # 初始化进程池，保留一个CPU核心避免系统卡顿
    pool = Pool(processes=multiprocessing.cpu_count()-1)
    # 分发任务并收集所有处理后的DataFrame
    results = pool.map(compile, file_list)
    pool.close()
    pool.join()

    # 合并结果并导出
    stock_list_merged = pd.concat(results)
    stock_list_merged.to_csv('scholes expanded.csv', index=False)

if __name__ == '__main__':
    main()

额外优化建议 #

• 如果CSV文件体积较大，可改用pool.imap或pool.imap_unordered，避免一次性加载所有结果到内存。
• 给pd.read_csv添加usecols参数，只加载需要的列（比如UnderlyingSymbol、Delta、OpenInterest），进一步提升读取速度。

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