如何有效地链接 ipyparallel 任务并将中间结果传递给引擎?
How to efficiently chain ipyparallel tasks and pass intermediate results to engines?
我正在尝试在 iPyParallel 中将多个任务链接在一起,例如
import ipyparallel
client = ipyparallel.Client()
view = client.load_balanced_view()
def task1(x):
## Do some work.
return x * 2
def task2(x):
## Do some work.
return x * 3
def task3(x):
## Do some work.
return x * 4
results1 = view.map_async(task1, [1, 2, 3])
results2 = view.map_async(task2, results1.get())
results3 = view.map_async(task3, results2.get())
但是,此代码不会提交任何任务 2,除非任务 1 已完成并且基本上处于阻塞状态。我的任务可能需要不同的时间,而且效率很低。 有没有一种简单的方法可以有效地链接这些步骤,并且引擎可以从前面的步骤中获取结果? 类似于:
def task2(x):
## Do some work.
return x.get() * 3 ## Get AsyncResult out.
def task3(x):
## Do some work.
return x.get() * 4 ## Get AsyncResult out.
results1 = [view.apply_async(task1, x) for x in [1, 2, 3]]
results2 = []
for x in result1:
view.set_flags(after=x.msg_ids)
results2.append(view.apply_async(task2, x))
results3 = []
for x in result2:
view.set_flags(after=x.msg_ids)
results3.append(view.apply_async(task3, x))
显然,这将失败,因为 AsyncResult 不可选取。
我正在考虑几个解决方案:
使用view.map_async(ordered=False)。
results1 = view.map_async(task1, [1, 2, 3], ordered=False)
for x in results1:
results2.append(view.apply_async(task2, x.get()))
但这必须等待所有task1 完成才能提交任何task3。还在阻塞。
使用异步。
@asyncio.coroutine
def submitter(x):
result1 = yield from asyncio.wrap_future(view.apply_async(task1, x))
result2 = yield from asyncio.wrap_future(view.apply_async(task2, result1)
result3 = yield from asyncio.wrap_future(view.apply_async(task3, result2)
yield result3
@asyncio.coroutine
def submit_all(ls):
jobs = [submitter(x) for x in ls]
results = []
for async_r in asyncio.as_completed(jobs):
r = yield from async_r
results.append(r)
## Do some work, like analysing results.
它工作正常,但是当引入更复杂的任务时,代码很快就会变得混乱和不直观。
感谢您的帮助。
选项一:链式期货
IPython parallel 在这方面不是最好的,因为连接必须在客户端级别完成。在提交结果之前,您必须等待结果完成并 return 给客户端。本质上,您的 asyncio submit_all 是实现 IPython 并行的正确方法。您可以通过编写一个 chain
函数来获得更通用的东西,该函数使用 add_done_callback
在前一个任务完成时提交新任务:
from concurrent.futures import Future
from functools import partial
def chain_apply(view, func, future):
"""Chain a call to view.apply(func, future.result()) when future is ready.
Returns a Future for the subsequent result.
"""
f2 = Future()
# when f1 is ready, submit a new task for func on its result
def apply_func(f):
if f.exception():
f2.set_exception(f.exception())
return
print('submitting %s(%s)' % (func.__name__, f.result()))
ar = view.apply_async(func, f.result())
# when ar is done, pass through the result to f2
ar.add_done_callback(lambda ar: f2.set_result(ar.get()))
future.add_done_callback(apply_func)
return f2
def chain_map(view, func, list_of_futures):
"""Chain a new callback on a list of futures."""
return [ chain_apply(view, func, f) for f in list_of_futures ]
# use builtin map with apply, since we want one Future per item
results1 = map(partial(view.apply, task1), [1, 2, 3])
results2 = chain_map(view, task2, results1)
results3 = chain_map(view, task3, results2)
print("Waiting for results")
[ r.result() for r in results3 ]
与 add_done_callback
的任何示例一样,它可以用协程编写,但我发现这种情况下的回调很好。这至少应该是一个相当通用的实用程序,您可以使用它来编写您的管道。
选项 2:dask.distributed
完全披露:我是 IPython Parallel 的主要作者,即将建议您使用不同的工具。
可以通过引擎名称空间和 DAG 依赖项并行 IPython 将结果从一个任务传递到另一个任务,但老实说,如果您的工作流程看起来像这样,您应该考虑使用 dask distributed,这是专门为这种计算图设计的。如果您已经习惯并熟悉 IPython parallel,那么开始使用 dask 应该不会是太大的负担。
IPython 5.1 提供了一个方便的命令,用于将您的 IPython 并行集群转变为 dask 分布式集群:
import ipyparallel as ipp
client = ipp.Client()
executor = client.become_distributed(ncores=1)
然后 dask 的关键相关特性是您可以将 futures 作为参数提交给后续的 map 调用,当结果准备好时调度程序会处理它,而不必在客户端中显式执行:
results1 = executor.map(task1, [1, 2, 3])
results2 = executor.map(task2, results1)
results3 = executor.map(task3, results2)
executor.gather(results3)
所以基本上,dask distributed 可以按照你希望的方式工作 IPython 当你需要像这样链接事情时,parallel 的负载平衡会工作。
This notebook 说明了这两个示例。
我正在尝试在 iPyParallel 中将多个任务链接在一起,例如
import ipyparallel
client = ipyparallel.Client()
view = client.load_balanced_view()
def task1(x):
## Do some work.
return x * 2
def task2(x):
## Do some work.
return x * 3
def task3(x):
## Do some work.
return x * 4
results1 = view.map_async(task1, [1, 2, 3])
results2 = view.map_async(task2, results1.get())
results3 = view.map_async(task3, results2.get())
但是,此代码不会提交任何任务 2,除非任务 1 已完成并且基本上处于阻塞状态。我的任务可能需要不同的时间,而且效率很低。 有没有一种简单的方法可以有效地链接这些步骤,并且引擎可以从前面的步骤中获取结果? 类似于:
def task2(x):
## Do some work.
return x.get() * 3 ## Get AsyncResult out.
def task3(x):
## Do some work.
return x.get() * 4 ## Get AsyncResult out.
results1 = [view.apply_async(task1, x) for x in [1, 2, 3]]
results2 = []
for x in result1:
view.set_flags(after=x.msg_ids)
results2.append(view.apply_async(task2, x))
results3 = []
for x in result2:
view.set_flags(after=x.msg_ids)
results3.append(view.apply_async(task3, x))
显然,这将失败,因为 AsyncResult 不可选取。
我正在考虑几个解决方案:
使用view.map_async(ordered=False)。
results1 = view.map_async(task1, [1, 2, 3], ordered=False) for x in results1: results2.append(view.apply_async(task2, x.get()))
但这必须等待所有task1 完成才能提交任何task3。还在阻塞。
使用异步。
@asyncio.coroutine def submitter(x): result1 = yield from asyncio.wrap_future(view.apply_async(task1, x)) result2 = yield from asyncio.wrap_future(view.apply_async(task2, result1) result3 = yield from asyncio.wrap_future(view.apply_async(task3, result2) yield result3 @asyncio.coroutine def submit_all(ls): jobs = [submitter(x) for x in ls] results = [] for async_r in asyncio.as_completed(jobs): r = yield from async_r results.append(r) ## Do some work, like analysing results.
它工作正常,但是当引入更复杂的任务时,代码很快就会变得混乱和不直观。
感谢您的帮助。
选项一:链式期货
IPython parallel 在这方面不是最好的,因为连接必须在客户端级别完成。在提交结果之前,您必须等待结果完成并 return 给客户端。本质上,您的 asyncio submit_all 是实现 IPython 并行的正确方法。您可以通过编写一个 chain
函数来获得更通用的东西,该函数使用 add_done_callback
在前一个任务完成时提交新任务:
from concurrent.futures import Future
from functools import partial
def chain_apply(view, func, future):
"""Chain a call to view.apply(func, future.result()) when future is ready.
Returns a Future for the subsequent result.
"""
f2 = Future()
# when f1 is ready, submit a new task for func on its result
def apply_func(f):
if f.exception():
f2.set_exception(f.exception())
return
print('submitting %s(%s)' % (func.__name__, f.result()))
ar = view.apply_async(func, f.result())
# when ar is done, pass through the result to f2
ar.add_done_callback(lambda ar: f2.set_result(ar.get()))
future.add_done_callback(apply_func)
return f2
def chain_map(view, func, list_of_futures):
"""Chain a new callback on a list of futures."""
return [ chain_apply(view, func, f) for f in list_of_futures ]
# use builtin map with apply, since we want one Future per item
results1 = map(partial(view.apply, task1), [1, 2, 3])
results2 = chain_map(view, task2, results1)
results3 = chain_map(view, task3, results2)
print("Waiting for results")
[ r.result() for r in results3 ]
与 add_done_callback
的任何示例一样,它可以用协程编写,但我发现这种情况下的回调很好。这至少应该是一个相当通用的实用程序,您可以使用它来编写您的管道。
选项 2:dask.distributed
完全披露:我是 IPython Parallel 的主要作者,即将建议您使用不同的工具。
可以通过引擎名称空间和 DAG 依赖项并行 IPython 将结果从一个任务传递到另一个任务,但老实说,如果您的工作流程看起来像这样,您应该考虑使用 dask distributed,这是专门为这种计算图设计的。如果您已经习惯并熟悉 IPython parallel,那么开始使用 dask 应该不会是太大的负担。
IPython 5.1 提供了一个方便的命令,用于将您的 IPython 并行集群转变为 dask 分布式集群:
import ipyparallel as ipp
client = ipp.Client()
executor = client.become_distributed(ncores=1)
然后 dask 的关键相关特性是您可以将 futures 作为参数提交给后续的 map 调用,当结果准备好时调度程序会处理它,而不必在客户端中显式执行:
results1 = executor.map(task1, [1, 2, 3])
results2 = executor.map(task2, results1)
results3 = executor.map(task3, results2)
executor.gather(results3)
所以基本上,dask distributed 可以按照你希望的方式工作 IPython 当你需要像这样链接事情时,parallel 的负载平衡会工作。
This notebook 说明了这两个示例。