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 ...

问题描述:

我正在使用 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:

第一个参数Valuetypecode_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。

相关推荐
  政府信创国产化的10大政策解读一、信创国产化的背景与意义信创国产化,即信息技术应用创新国产化,是当前中国信息技术领域的一个重要发展方向。其核心在于通过自主研发和创新,实现信息技术应用的自主可控,减少对外部技术的依赖,并规避潜在的技术制裁和风险。随着全球信息技术竞争的加剧,以及某些国家对中国在科技领域的打压,信创国产化显...
工程项目管理   1603  
  为什么项目管理通常仍然耗时且低效?您是否还在反复更新电子表格、淹没在便利贴中并参加每周更新会议?这确实是耗费时间和精力。借助软件工具的帮助,您可以一目了然地全面了解您的项目。如今,国内外有足够多优秀的项目管理软件可以帮助您掌控每个项目。什么是项目管理软件?项目管理软件是广泛行业用于项目规划、资源分配和调度的软件。它使项...
项目管理软件   1369  
  信创产品在政府采购中的占比分析随着信息技术的飞速发展以及国家对信息安全重视程度的不断提高,信创产业应运而生并迅速崛起。信创,即信息技术应用创新,旨在实现信息技术领域的自主可控,减少对国外技术的依赖,保障国家信息安全。政府采购作为推动信创产业发展的重要力量,其对信创产品的采购占比情况备受关注。这不仅关系到信创产业的发展前...
信创和国产化的区别   30  
  信创,即信息技术应用创新产业,旨在实现信息技术领域的自主可控,摆脱对国外技术的依赖。近年来,国货国用信创发展势头迅猛,在诸多领域取得了显著成果。这一发展趋势对科技创新产生了深远的推动作用,不仅提升了我国在信息技术领域的自主创新能力,还为经济社会的数字化转型提供了坚实支撑。信创推动核心技术突破信创产业的发展促使企业和科研...
信创工作   28  
  信创技术,即信息技术应用创新产业,旨在实现信息技术领域的自主可控与安全可靠。近年来,信创技术发展迅猛,对中小企业产生了深远的影响,带来了诸多不可忽视的价值。在数字化转型的浪潮中,中小企业面临着激烈的市场竞争和复杂多变的环境,信创技术的出现为它们提供了新的发展机遇和支撑。信创技术对中小企业的影响技术架构变革信创技术促使中...
信创国产化   35  
热门文章
项目管理软件有哪些?
云禅道AD
禅道项目管理软件

云端的项目管理软件

尊享禅道项目软件收费版功能

无需维护,随时随地协同办公

内置subversion和git源码管理

每天备份,随时转为私有部署

免费试用