具有依赖于模块的嵌套函数的并行映射

Question

这是我尝试

import numpy as np

def generate_function(a):
    def func(x):
        '''a complex function that uses several modules'''
        return x + np.sqrt(a)
    return func

if __name__ == '__main__':
    f = generate_function(0.5)
    x = np.arange(0, 100)
    y = np.array(list(map(f, x))) # want to parallelize this step

与 multiprocessing 并行化的一个最小示例，嵌套的 func 会导致问题，因为 pickle 无法访问嵌套函数

import multiprocessing as mp
...
pool = mp.Pool(2)
y = np.array(pool.map(f, x))

AttributeError: Can't pickle local object 'generate_function.<locals>.func'

即使使用 pathos，模块也不会导入

import pathos
...
pool = pathos.multiprocessing.ProcessPool(2)
y = np.array(pool.map(f, x))

NameError: name 'np' is not defined

请注意，Python 多处理 PicklingError：无法 pickle 要么工作

并行化的最佳方法是什么？

---

因此，通过在 generate_function 内部重新导入来使 pathos 工作是可能的

def generate_function(a):
    import numpy as np
    def func(x):
        '''a complex function that uses several modules'''
        return x + np.sqrt(a)
    return func

，但我可能有多个导入多个 generate_function > 和多层嵌套，跟踪所有这些很快就会变得非常麻烦，所以我想避免这种混乱

def generate_function1(a):
    import module1, module2, module3
    from module4 import a, b
    from module5 import c as d
    from module6 import e as f
    def func(x):
        ...
    return func

def generate_function2(a):
    import module1, module2, module3
    from module4 import a, b
    from module5 import c as d
    from module6 import e as f
    def func(x):
        ...
    return func

def generate_generator_function(a):
    import module1, module2, module3
    from module4 import a, b
    from module5 import c as d
    from module6 import e as f
    def generate_function(a):
        import module1, module2, module3
        from module4 import a, b
        from module5 import c as d
        from module6 import e as f
        def func(x):
            ...
        return func
    return generate_function

Answer 1

您可以使用concurrent.futures：

import concurrent.futures

f = generate_function(0.5)
x = np.arange(0, 100)
with concurrent.futures.ThreadPoolExecutor() as ex:
    y = ex.map(f, x)

Answer 2

这不会解决您的泡菜问题，但这是我对管理导入的类的想法。

>>> class funcFactory:
...     import numpy as np
...     def __init__(self):
...             pass
...     def makef(self,a):
...             def func(x):
...                     return a+funcFactory.np.sqrt(x)
...             return func
...
>>> ff = funcFactory()
>>> f = ff.makef(1)
>>> f(4)
3.0

结合@Schotty关于使用concurrent.futures的建议，你最终会得到如下所示的代码：

import concurrent.futures
import numpy as np

class funcFactory:
    import numpy as np
    def makef(self,a):
        def func(x):
            return a+funcFactory.np.sqrt(x)
        func.__reduce__ = lambda:""
        return func

f = funcFactory().makef(0.5)
with concurrent.futures.ThreadPoolExecutor() as ex:
    y = ex.map(f, np.arange(0, 100))
print(list(y))