python 中的多处理-在多个进程之间共享大对象（例如 pandas 数据框）
- 2025-03-14 08:57:00
- admin 原创
- 14
问题描述：
我正在使用 Python 多处理，更准确地说
from multiprocessing import Pool
p = Pool(15)
args = [(df, config1), (df, config2), ...] #list of args - df is the same object in each tuple
res = p.map_async(func, args) #func is some arbitrary function
p.close()
p.join()
这种方法的内存消耗很大；几乎占用了我所有的内存（此时它变得非常慢，因此使多处理变得毫无用处）。我猜问题是这df是一个巨大的对象（一个大的 pandas 数据框），并且它会被复制到每个进程。我曾尝试使用multiprocessing.Value共享数据框而不进行复制
shared_df = multiprocessing.Value(pandas.DataFrame, df)
args = [(shared_df, config1), (shared_df, config2), ...]
（如Python 多处理共享内存中所建议的那样），但这给了我（与在 Python 进程之间共享复杂对象TypeError: this type has no size相同？ ，不幸的是我不明白答案）。
我第一次使用多处理，可能我的理解还不够好。multiprocessing.Value在这种情况下，这真的是正确的选择吗？我看过其他建议（例如队列），但现在有点困惑。有哪些选项可以共享内存，在这种情况下哪一个最好？
解决方案 1：
第一个参数Value是typecode_or_type。其定义如下：
typecode_or_type 确定返回对象的类型：它要么是 ctypes 类型，要么是 array 模块使用的类型的一个字符类型代码。 *args 被传递给该类型的构造函数。
重点是我。因此，您根本无法将 pandas 数据框放入 中Value，它必须是ctypes 类型。
您也可以使用multiprocessing.Manager为所有进程提供单例数据框实例。有几种不同的方法可以达到相同的效果 - 最简单的方法可能是将您的数据框放入管理器的 中Namespace。
from multiprocessing import Manager
mgr = Manager()
ns = mgr.Namespace()
ns.df = my_dataframe
# now just give your processes access to ns, i.e. most simply
# p = Process(target=worker, args=(ns, work_unit))
现在，任何向 Manager 传递引用的进程都可以访问您的数据框实例。或者，只需向 传递引用Namespace，这样更简洁。
我没有/不会涉及的一件事是事件和信号 - 如果您的进程需要等待其他进程完成执行，则需要添加它。 这里有一个包含一些示例的页面Event，其中还更详细地介绍了如何使用管理器Namespace。
（请注意，这些都不能说明是否multiprocessing会带来切实的性能优势，而只是为您提供了探索该问题的工具）
解决方案 2：
您可以使用Array而不是Value存储数据框。
以下解决方案将pandas数据框转换为将其数据存储在共享内存中的对象：
import numpy as np
import pandas as pd
import multiprocessing as mp
import ctypes
# the origingal dataframe is df, store the columns/dtypes pairs
df_dtypes_dict = dict(list(zip(df.columns, df.dtypes)))
# declare a shared Array with data from df
mparr = mp.Array(ctypes.c_double, df.values.reshape(-1))
# create a new df based on the shared array
df_shared = pd.DataFrame(np.frombuffer(mparr.get_obj()).reshape(df.shape),
columns=df.columns).astype(df_dtypes_dict)
如果您现在df_shared跨进程共享，则不会进行额外复制。对于您的情况：
pool = mp.Pool(15)
def fun(config):
# df_shared is global to the script
df_shared.apply(config) # whatever compute you do with df/config
config_list = [config1, config2]
res = p.map_async(fun, config_list)
p.close()
p.join()
如果你使用pandarallel这也特别有用，例如：
# this will not explode in memory
from pandarallel import pandarallel
pandarallel.initialize()
df_shared.parallel_apply(your_fun, axis=1)
注意：使用此解决方案，您最终会得到两个数据框（df 和 df_shared），它们消耗两倍的内存，并且初始化时间很长。可能可以直接在共享内存中读取数据。
解决方案 3：
通过创建 data_handler 子进程，您可以在进程之间共享 pandas 数据框，而无需任何内存开销。此进程接收来自其他子进程的调用，这些子进程对非常大的数据框对象具有特定的数据请求（即一行、特定单元格、切片等）。只有 data_handler 进程会将您的数据框保存在内存中，而不像 Namespace 之类的管理器那样会将数据框复制到所有子进程。请参阅下面的工作示例。这可以转换为池。
需要进度条吗？请参阅我的回答：https://stackoverflow.com/a/55305714/11186769
import time
import Queue
import numpy as np
import pandas as pd
import multiprocessing
from random import randint
#==========================================================
# DATA HANDLER
#==========================================================
def data_handler( queue_c, queue_r, queue_d, n_processes ):
# Create a big dataframe
big_df = pd.DataFrame(np.random.randint(
0,100,size=(100, 4)), columns=list('ABCD'))
# Handle data requests
finished = 0
while finished < n_processes:
try:
# Get the index we sent in
idx = queue_c.get(False)
except Queue.Empty:
continue
else:
if idx == 'finished':
finished += 1
else:
try:
# Use the big_df here!
B_data = big_df.loc[ idx, 'B' ]
# Send back some data
queue_r.put(B_data)
except:
pass
# big_df may need to be deleted at the end.
#import gc; del big_df; gc.collect()
#==========================================================
# PROCESS DATA
#==========================================================
def process_data( queue_c, queue_r, queue_d):
data = []
# Save computer memory with a generator
generator = ( randint(0,x) for x in range(100) )
for g in generator:
"""
Lets make a request by sending
in the index of the data we want.
Keep in mind you may receive another
child processes return call, which is
fine if order isnt important.
"""
#print(g)
# Send an index value
queue_c.put(g)
# Handle the return call
while True:
try:
return_call = queue_r.get(False)
except Queue.Empty:
continue
else:
data.append(return_call)
break
queue_c.put('finished')
queue_d.put(data)
#==========================================================
# START MULTIPROCESSING
#==========================================================
def multiprocess( n_processes ):
combined = []
processes = []
# Create queues
queue_data = multiprocessing.Queue()
queue_call = multiprocessing.Queue()
queue_receive = multiprocessing.Queue()
for process in range(n_processes):
if process == 0:
# Load your data_handler once here
p = multiprocessing.Process(target = data_handler,
args=(queue_call, queue_receive, queue_data, n_processes))
processes.append(p)
p.start()
p = multiprocessing.Process(target = process_data,
args=(queue_call, queue_receive, queue_data))
processes.append(p)
p.start()
for i in range(n_processes):
data_list = queue_data.get()
combined += data_list
for p in processes:
p.join()
# Your B values
print(combined)
if __name__ == "__main__":
multiprocess( n_processes = 4 )
解决方案 4：
至少Python 3.6支持将 pandas DataFrame 存储为 multiprocessing.Value。请参阅以下工作示例：
import ctypes
import pandas as pd
from multiprocessing import Value
df = pd.DataFrame({'a': range(0,9),
'b': range(10,19),
'c': range(100,109)})
k = Value(ctypes.py_object)
k.value = df
print(k.value)
解决方案 5：
我很惊讶joblib 的 Parallel（至少从 1.0.1 开始）已经支持开箱即用地与多进程工作程序共享 pandas 数据框。至少使用“loky”后端。我通过实验得出了一个结论：传递给函数的参数不应包含任何大型字典。如果包含，请将字典转换为 Series 或 Dataframe。每个工作程序肯定会使用一些额外的内存，但比主进程中所谓的“大”数据框的大小要少得多。并且所有工作程序都会立即开始计算。否则，joblib 会启动您请求的所有工作程序，但它们会处于空闲状态，而对象会按顺序复制到每个工作程序中，这需要很长时间。如果有人需要，我可以提供代码示例。我已经测试了仅以只读模式处理数据框。文档中没有提到该功能，但它适用于 Pandas。
