python - 使用多处理时实例属性不会持续存在

Question

我遇到的问题是实例不保留对属性的更改，甚至不保留创建的新属性。我认为我已经将范围缩小到我的脚本利用了多处理这一事实，并且我认为当脚本返回到主线程时，单独进程线程中的实例发生的更改不会被“记住”。

基本上，我有几组数据需要并行处理。数据作为属性存储，并通过类中的多个方法进行更改。处理结束时，我希望返回主线程并连接每个对象实例的数据。但是，如上所述，当我在并行处理位完成后尝试使用数据访问实例属性时，那里什么也没有。就好像多处理期间发生的任何更改都被“遗忘”了。

有没有明显的解决方案来解决这个问题？或者我是否需要重建代码以返回处理后的数据，而不仅仅是将其更改/存储为实例属性？我想另一种解决方案是序列化数据，然后在必要时重新读取它，而不是仅仅将其保留在内存中。

这里可能值得注意的是，我使用的是 pathos 模块，而不是 python 的 multiprocessing 模块。我收到一些与酸洗有关的错误，类似于这里:Python multiprocessing PicklingError: Can't pickle <type 'function'> 。我的代码分为多个模块，如上所述，数据处理方法包含在一个类中。

抱歉文字墙。

编辑 这是我的代码:

import importlib
import pandas as pd
from pathos.helpers import mp
from provider import Provider

# list of data providers ... length is arbitrary
operating_providers = ['dataprovider1', 'dataprovider2', 'dataprovider3']


# create provider objects for each operating provider
provider_obj_list = []
for name in operating_providers:
    loc     = 'providers.%s' % name
    module  = importlib.import_module(loc)
    provider_obj = Provider(module)
    provider_obj_list.append(provider_obj)

processes = []
for instance in provider_obj_list:
    process = mp.Process(target = instance.data_processing_func)
    process.daemon = True
    process.start()
    processes.append(process)

for process in processes:
    process.join()

# now that data_processing_func is complete for each set of data, 
# stack all the data
stack = pd.concat((instance.data for instance in provider_obj_list))

我有许多模块(它们的名称在 operating_providers 中列出)，其中包含特定于其数据源的属性。这些模块被迭代地导入并传递给 Provider 类的新实例，该类是我在单独的模块 (provider) 中创建的。我将每个 Provider 实例附加到一个列表 (provider_obj_list)，然后迭代创建调用实例方法 instance.data_processing_func 的单独进程。该函数执行一些数据处理(每个实例访问完全不同的数据文件)，并在此过程中创建新的实例属性，当并行处理完成时我需要访问这些属性。

我尝试使用多线程，而不是多处理——在这种情况下，我的实例属性仍然存在，这正是我想要的。但是，我不确定为什么会发生这种情况 - 我必须研究线程与多处理之间的差异。

感谢您的帮助!

Answer 1

这里有一些示例代码，展示了如何执行我在评论中概述的操作。我无法测试它，因为我没有安装 provider 或 pathos，但它应该能让您很好地了解我的建议。

import importlib
from pathos.helpers import mp
from provider import Provider

def process_data(loc):
    module  = importlib.import_module(loc)
    provider_obj = Provider(module)
    provider_obj.data_processing_func()


if __name__ == '__main__':

    # list of data providers ... length is arbitrary
    operating_providers = ['dataprovider1', 'dataprovider2', 'dataprovider3']

    # create list of provider locations for each operating provider
    provider_loc_list = []
    for name in operating_providers:
        loc = 'providers.%s' % name
        provider_loc_list.append(loc)

    processes = []
    for loc in provider_loc_list:
        process = mp.Process(target=process_data, args=(loc,))
        process.daemon = True
        process.start()
        processes.append(process)

    for process in processes:
        process.join()