python 中的多处理-在多个进程之间共享大对象(例如 pandas 数据框)
- 2025-03-14 08:57:00
- admin 原创
- 12
问题描述:
我正在使用 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。