任务上的进程池

Question

我是DASK的新手。

我可以使用client.map(funct_name，迭代器)提交10个任务，其中迭代器是一个包含10个元素的列表。

现在，当之前提交的10个任务中的任何一个任务完成时，我想提交下一个任务，比如说第11个任务。

我知道在python中有一种叫做进程池的东西。但是我想使用DASK来实现类似于进程池的东西。

请有人指导我使用DASK进程池。

Answer 1

您可以做的最简单的事情是在提交新工作之前使用wait

futs = client.map(funct_name, iterator)
distributed.wait(futs)
out = client.submit(eleventh, args)

然而，如果您想在前十项工作完成时提交您的新工作，但让它自动等待，直到所有工作都完成，您可以构造一个表面上依赖于前一项工作，但实际上并不利用它们的假任务

futs = client.map(funct_name, iterator)

def run_eleventh(args, deps):
    return eleventh(args)

out = client.submit(run_eleventh, (args, futs))

Answer 2

您可能希望在此处查看as_completed对象：

http://docs.dask.org/en/latest/futures.html#waiting-on-futures

from dask.distributed import as_completed

futures = client.map(score, x_values)

for future in as_completed(futures):
    ...
    client.submit(...)