Jupyter Notebook 永远不会使用多处理完成处理（Python 3）

Question

木星笔记本

我基本上使用多处理模块，我仍在学习多处理的功能。我正在使用 Dusty Phillips 的书，这段代码属于它。

 import multiprocessing
import random
from multiprocessing.pool import Pool

def prime_factor(value):
    factors = []
    for divisor in range(2, value-1):
        quotient, remainder = divmod(value, divisor)
        if not remainder:
            factors.extend(prime_factor(divisor))
            factors.extend(prime_factor(quotient))
            break
        else:
            factors = [value]
    return factors

if __name__ == '__main__':
    pool = Pool()
    to_factor = [ random.randint(100000, 50000000) for i in range(20)]
    results = pool.map(prime_factor, to_factor)
    for value, factors in zip(to_factor, results):
        print("The factors of {} are {}".format(value, factors))

在 Windows PowerShell 上（不是在 jupyter notebook 上）我看到以下内容

Process SpawnPoolWorker-5:
Process SpawnPoolWorker-1:
AttributeError: Can't get attribute 'prime_factor' on <module '__main__' (built-in)>

我不知道为什么单元格永远不会停止运行？

原文由 rsc05 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

似乎 Jupyter notebook 和不同 ide 中的问题是设计特征。因此，我们必须将函数 (prime_factor) 写入不同的文件并导入模块。此外，我们必须注意调整。例如，在我的例子中，我将函数编码到一个名为 defs.py 的文件中

def prime_factor(value):
    factors = []
    for divisor in range(2, value-1):
        quotient, remainder = divmod(value, divisor)
        if not remainder:
            factors.extend(prime_factor(divisor))
            factors.extend(prime_factor(quotient))
            break
        else:
            factors = [value]
    return factors

然后在 jupyter notebook 中我写了以下几行

import multiprocessing
import random
from multiprocessing import Pool
import defs

if __name__ == '__main__':
    pool = Pool()
    to_factor = [ random.randint(100000, 50000000) for i in range(20)]
    results = pool.map(defs.prime_factor, to_factor)
    for value, factors in zip(to_factor, results):
        print("The factors of {} are {}".format(value, factors))

这解决了我的问题

原文由 rsc05 发布，翻译遵循 CC BY-SA 3.0 许可协议

Answer 2

要执行函数 而不必手动将其写入单独的文件：

我们可以将要处理的任务动态写入临时文件，导入并执行函数。

 from multiprocessing import Pool
from functools import partial
import inspect

def parallel_task(func, iterable, *params):

    with open(f'./tmp_func.py', 'w') as file:
        file.write(inspect.getsource(func).replace(func.__name__, "task"))

    from tmp_func import task

    if __name__ == '__main__':
        func = partial(task, params)
        pool = Pool(processes=8)
        res = pool.map(func, iterable)
        pool.close()
        return res
    else:
        raise "Not in Jupyter Notebook"

然后我们可以像这样在笔记本单元中简单地调用它：

 def long_running_task(params, id):
    # Heavy job here
    return params, id

data_list = range(8)

for res in parallel_task(long_running_task, data_list, "a", 1, "b"):
    print(res)

输出：

 ('a', 1, 'b') 0
('a', 1, 'b') 1
('a', 1, 'b') 2
('a', 1, 'b') 3
('a', 1, 'b') 4
('a', 1, 'b') 5
('a', 1, 'b') 6
('a', 1, 'b') 7

注意：如果您使用的是 Anaconda，并且想查看繁重任务的进度，则可以在 — 中使用 print() long_running_task() 。打印的内容将显示在 Anaconda Prompt 控制台中。

原文由 H4dr1en 发布，翻译遵循 CC BY-SA 4.0 许可协议