我基本上使用的是多处理模块,我仍在学习多处理的功能。我正在使用达斯蒂·菲利普斯的书,这个代码属于它。
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)>
我不知道为什么单元格永远不会结束运行?
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笔记本中我写了以下几行
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))
这解决了我的问题
执行函数无需手动将其写入单独的文件:
我们可以动态地将要处理的任务写入临时文件中,导入并执行函数。
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 控制台中。
严格来说,即使添加了
if __name__="__main__"
,Windows Jupyter Notebook 也不支持 Python 多处理。
Windows 10 中的一种解决方法是将 Windows 浏览器与 WSL 中的 Jupyter 服务器连接。
您可以获得与Linux相同的体验。
您可以手动设置或参考https://github.com/mszhanyi/gemini
中的脚本另一个选择:使用 dask,它与 Jupyter 配合得很好。即使您不需要任何 dask 特殊数据结构,您也可以简单地使用它来控制多个进程。
为了处理让多进程在 Jupyter 会话中正常运行的许多怪癖,我创建了一个库
mpify
,它允许一次性执行多进程函数,并使用简单的 API 将内容从笔记本传递到子进程。
Jupyter shell进程本身可以作为工作进程参与。 用户可以选择收集所有工作人员的结果,或仅收集其中一名工作人员的结果。
这是:
https://github.com/philtrade/mpify
在底层,它使用
multiprocess
—— 来自标准 python multiprocessing
库的一个积极支持的分支 —— 允许在子进程中访问笔记本中本地定义的变量/函数。 它还使用 spawn
启动方法,如果子进程要使用多个 GPU(这是一个越来越常见的用例),则该方法是必需的。 它使用 Process()
而不是 Pool()
,来自 multiprocess
API。
用户可以提供自定义上下文管理器来获取资源,设置/拆除函数执行周围的执行环境。 我提供了一个示例上下文管理器来支持 PyTorch 的分布式数据并行 (DDP) 设置,以及更多有关如何使用 DDP 在多个 GPU 上的 Jupyter 中训练
fastai v2
的示例。
欢迎分享错误报告、PR、用例。
绝不是一个花哨/强大的库,
mpify
仅打算支持单主机/多进程类型的分布式设置,并且只是生成-执行-终止。它也不支持持久进程池和花哨的任务调度——ipyparallel
或dask
已经做到了。
我希望它对那些正在努力使用 Jupyter + 多处理以及多 GPU 的人们有用。 谢谢。
只需将文件另存为 .py 然后使用 jupyter 终端 -> 进入你的 name.py 目录 -> python name.py 它适用于所有多处理情况