Python中的多进程 - 在多个进程之间共享大型对象（例如pandas dataframe）

Question

我正在使用Python multiprocessing，更准确地说

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 dataframe），每个进程都要复制一份。我尝试使用multiprocessing.Value来共享数据框而不进行复制。

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

根据 Python multiprocessing shared memory 建议，我使用了 multiprocessing.Value，但是出现了 TypeError: this type has no size 错误（与 Sharing a complex object between Python processes? 相同，不幸的是，我不理解这个答案）。 这是我第一次使用多进程，也许我的理解还不够好。在这种情况下，multiprocessing.Value 真的是正确的使用方式吗？我已经看到了其他建议（例如队列），但现在有点困惑。有哪些共享内存的选项，在这种情况下哪一个是最好的？

- Anne

---

请参考最近相关问题：http://stackoverflow.com/questions/22468279/slow-speed-while-parallelizing-operation-on-pandas-dataframe。 - Jeff

有最近的方法来做这件事吗，还是使用Namespace仍然是最好的方法？你最终是如何解决的@Anne - David

Answer 1

Value的第一个参数是typecode_or_type，它被定义为：

typecode_or_type确定返回对象的类型：它可以是ctypes类型或由数组模块使用的单个字符类型代码。*args被传递给类型的构造函数。

重点在于这里。因此，您不能将pandas dataframe放入Value中，它必须是ctypes类型。

相反，您可以使用multiprocessing.Manager将单例dataframe实例提供给所有进程。有几种不同的方法可以到达相同的位置——可能最简单的方法就是将dataframe放入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))

现在您的数据帧实例可以被任何传递给管理器引用的进程访问。或者只需传递对“名称空间”的引用，这更加清晰。

我没有/不会涉及事件和信号-如果您的进程需要等待其他进程完成执行，则需要添加它。这里是一个页面，其中包含一些 Event 示例，还详细介绍了如何使用管理器的 Namespace 。

（请注意，所有这些都没有解决 multiprocessing 是否会产生有形的性能收益的问题，这只是为您提供探索该问题的工具）

- roippi

---

谢谢，这使得内存消耗有了很大的改善。不过它仍比我想象中要高得多 - 我该如何找出内存消耗来自哪里？ - Anne

@Anne，我也发现这种“命名空间”方法会导致很多内存消耗。我尝试过使用具有数百万行和 6 列（占用 2 GB RAM）的 DF 进行操作，工作线程最终也使用了差不多这么多的内存。更糟糕的是，在多进程上下文中，对于在非多进程环境下进行分析时快速（<1 毫秒）的数据访问，在工作进程中变得非常缓慢。甚至在工作进程的内存使用量增加后，单个 ns.df.loc[ix] 调用可能需要几秒钟的时间。@roippi 和 @Jeff，你们对此有什么想法吗？ - Scott H

我尝试使用一个大的数据框（从一个约9GB的csv文件中加载）来尝试这种方法，但是我遇到了一个格式化错误，我猜测这是由于数据框的大小所致，"struct.error: 'i' format requires -2147483648 <= number <= 2147483647"，你有什么建议吗？是否有其他可用的技术？ - p1nox

如果我们创建了两个“进程”或者一个“进程池”，那么在“命名空间”内的任何内容是否会被复制到不同的子进程中？我很困惑，也很想知道是否使用multiprocessing.Manager可以避免巨大的内存消耗。 - Sta_Doc

这不是在进程之间共享数据帧的解决方案。相关主题：https://stackoverflow.com/questions/72798554/how-to-use-multiprocessing-to-share-a-large-database-among-processes/72817277#72817277 - undefined

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

- toine

---

这似乎是唯一可行的共享pandas dataframe而不需要复制到每个子进程，同时能够使用多核计算的方法。MP中的其他东西（如命名空间和管理器）仍会创建副本。只有使用Array时，主进程的内存使用量才会从原始df的7倍（在共享时）降至3倍（在运行时共享后，至少在Windows上），并且重新创建df需要显着的时间。是否有任何方式可以进一步优化/加速这个过程？ - Anatoly Alekseev

Answer 3

至少 Python 3.6 支持将 Pandas DataFrame 存储为多进程值。以下是一个可行的示例：

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)

- artemisian

---

即使完成了，我的RAM使用量也增加了。 - Joao Vitor Deon

我试了一下。它在将Value对象作为参数传递给Process时有效，但如果我尝试从子进程中提取值，它总是崩溃。这段代码如果有一个完整的示例会更好。 - Anthony Nash

Answer 4

您可以通过创建一个数据处理器子进程，在进程间不产生任何内存开销地共享pandas dataframe。此进程接收来自其他子进程的特定数据请求（例如行、特定单元格、切片等）从您的非常大的dataframe对象中。只有data_handler进程保留您的dataframe在内存中，而不像Manager Namespace一样导致将dataframe复制到所有子进程中。以下是一个可工作的示例。这可以转换为池。

需要一个进度条吗？请参见我的答案：https://dev59.com/U2035IYBdhLWcg3wBLXb#55305714

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 )

- user11186769

---

嘿，感谢分享这个。它似乎正在工作。关于这个的一个小问题，数据处理程序中的队列是否会使进程处理程序逐个运行？在计算结果之前，进程处理程序只是等待，对吗？我们难道不是失去了并行化的主要目的吗？如果我在这里错了，请纠正我。谢谢！ - Pradeep Kumar Nalluri

Answer 5

我很惊讶地发现，自 1.0.1 版本以来，joblib 的 Parallel 已经支持使用 'loky' 后端将 pandas 数据框与多进程工作进程共享。但是需要注意的是，你传递给函数的参数不应该包含任何大型字典。如果有，需要将字典转换为 Series 或 Dataframe。每个工作进程肯定会使用一些额外的内存，但是比起主进程中所谓的“大”数据框大小，要少得多。而且所有工作进程都会立即开始计算。否则，joblib 会启动所有请求的工作进程，但它们会空闲挂起，直到对象被顺序复制到每个进程中，这需要很长时间。如果需要，我可以提供代码示例。我已在只读模式下测试了数据框处理。虽然此功能未在文档中提到，但它对 Pandas 可用。

- Anatoly Alekseev

---

一个代码示例会很棒！我从来没有让loky在我的问题上起作用。 - xan

代码示例在这里：https://github.com/joblib/joblib/issues/1244 - Anatoly Alekseev