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

这种方法消耗了大量内存，几乎把我的所有内存都用光了（到那时速度变得非常慢，因此多进程的效果几乎没有用）。我猜问题在于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

我挺惊讶的，joblib的Parallel（至少从1.0.1版本开始）已经可以直接支持在多进程工作中共享pandas数据框了。至少在使用'loky'这个后端时是这样的。
我通过实验发现，有一点需要注意：你传给函数的参数里不要包含大的字典。如果有的话，最好把字典转换成Series或DataFrame。
每个工作进程确实会多占用一些内存，但比起你在主进程中的那个“大的”数据框来说，使用的内存要少得多。而且所有的工作进程会立刻开始计算。如果不这样做，joblib会启动你请求的所有工作进程，但它们会在那儿闲着，等着把对象一个一个复制过去，这样会花费很长时间。如果有人需要，我可以提供代码示例。我只测试了数据框的只读模式。这个功能在文档里没有提到，但在Pandas中是可以用的。

Answer 2

至少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)

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

Answer 4

你可以用 Array 来代替 Value 来存储你的数据框（dataframe）。

下面的解决方案将一个 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 5

在Value的第一个参数是typecode_or_type。这个参数的意思是：

typecode_or_type决定了返回对象的类型：它可以是ctypes类型，或者是数组模块使用的一种单字符类型代码。 *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))

现在，任何获得管理器引用的进程都可以访问你的数据框实例。或者直接传递Namespace的引用，这样更简洁。

有一点我没有提到的就是事件和信号——如果你的进程需要等待其他进程完成执行，你需要添加这些功能。这里有一页提供了一些Event的例子，并且更详细地介绍了如何使用管理器的Namespace。

（注意，这里并没有讨论multiprocessing是否会带来实际的性能提升，这只是给你一些工具，让你可以去探索这个问题。）