python中的多处理 - 在多个进程之间共享大对象(例如pandas dataframe)

Question

我更精确地使用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()

这种方法具有巨大的内存消耗; 几乎占用了我所有的RAM(此时它变得非常慢,因此使多处理非常无用).我假设问题是这df是一个巨大的对象(一个大型的pandas数据帧),它会被复制到每个进程.我试过使用multiprocessing.Value共享数据帧而不复制

shared_df = multiprocessing.Value(pandas.DataFrame, df)
args = [(shared_df, config1), (shared_df, config2), ...] 

(正如Python多处理共享内存中所建议的那样),但是这给了我TypeError: this type has no size(与在Python进程之间共享一个复杂对象相同？,遗憾的是我不理解答案).

我第一次使用多处理,也许我的理解还不够好.是multiprocessing.Value实际上即使在这种情况下使用了正确的事情？我已经看到了其他建议(例如队列),但现在有点困惑.有什么选择可以共享内存,在这种情况下哪一个最好？

Answer 1

第一个参数Value是typecode_or_type.这被定义为:

typecode_or_type确定返回对象的类型:它是一个ctypes类型或一个数组模块使用的类型的一个字符类型代码.*args被传递给该类型的构造函数.

强调我的.所以,你根本不能把一个pandas数据帧放在一个Value,它必须是一个ctypes类型.

您可以改为使用a 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会带来切实的性能优势,这只是为您提供探索该问题的工具)

Answer 2

您可以通过创建 data_handler 子进程在进程之间共享 pandas 数据帧，而无需任何内存开销。该进程从其他子进程接收来自非常大的数据帧对象的特定数据请求（即行、特定单元格、切片等......）的调用。只有 data_handler 进程将数据帧保留在内存中，这与命名空间之类的管理器不同，后者会导致数据帧被复制到所有子进程。请参阅下面的工作示例。这可以转换为池。

需要一个进度条吗？在这里查看我的答案：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 )

Answer 3

您可以使用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），它们消耗两倍的内存并且初始化时间很长。可以直接在共享内存中读取数据。

Answer 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)