代码之家 › 专栏 › 技术社区 › rsc05

Jupyter notebook从不使用多处理完成处理(Python 3)

jupyter multiprocessing debugging python-3.x

rsc05 · 技术社区 · 8 年前

Jupyter笔记本

我基本上在使用多处理模块,我还在学习多处理的能力。我正在使用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笔记本)上,我看到了以下内容

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

我不知道为什么牢房一直在运行?

5 回复 | 直到 8 年前

rsc05 8 年前

在不同的ide中,Jupyter笔记本的问题似乎是设计特性。因此,我们必须将函数(prime_因子)写入另一个文件并导入模块。此外,我们必须注意调整。例如,在我的例子中,我将函数编码到一个名为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

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

这解决了我的问题

jtlz2 3 年前

执行函数

我们可以动态地将要处理的任务写入一个临时文件,导入它并执行该函数。

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提示控制台中。

Yi Zhang 4 年前

严格来说,Windows Jupyter笔记本甚至不支持Python多处理 if __name__="__main__" 已添加。

Windows 10中的一个解决方法是将Windows浏览器与WSL中的Jupyter服务器连接。

您可以获得与Linux相同的体验。

您可以手动设置或在中引用脚本 https://github.com/mszhanyi/gemini

user13422231 5 年前

为了解决让多进程在Jupyter会话中正常运行的许多怪癖,我创建了一个库 mpify 它允许一次性执行多进程函数,并使用简单的API将内容从笔记本传递到子进程。

Jupyter shell进程本身可以作为辅助进程参与。用户可以选择从所有工作人员收集结果,也可以仅从其中一个工作人员收集结果。

这是:

https://github.com/philtrade/mpify

在引擎盖下,它使用 multiprocess --来自标准python的受积极支持的fork multiprocessing spawn start方法,如果子流程要使用多个GPU,这是必要的,这是一种越来越常见的用例。它使用 Process() 不 Pool() ,从 多进程

用户可以提供自定义上下文管理器来获取资源,设置/拆除围绕函数执行的执行环境。我提供了一个示例上下文管理器来支持PyTorch的分布式数据并行(DDP)设置,以及更多关于如何训练的示例 fastai v2 在Jupyter中,使用DDP在多个GPU上运行。

欢迎分享错误报告、PRs和用例。

绝不是一个奇特/强大的图书馆, mpify公司 ipyparallel 或 dask 已经做到了。

我希望它能对那些正在努力使用Jupyter+多处理的人们有用,也可能对多GPU有用。谢谢

meizy 3 年前

另一种选择:使用dask,它可以很好地与Jupyter配合使用。即使您不需要任何dask特殊数据结构,也可以简单地使用它来控制多个进程。