1

我需要工人一次处理一个任务，并在开始新的任务之前完成当前流程。我无法做到：（1）每个工人在任何时候最多运行一个任务，（2）让工人在开始新的程序之前完成一个程序；原子事务。

我在具有 40 个节点的集群上使用 dask.distributed Client；4 核和 15GB 内存。我处理的管道有大约 8-10GB 的任务，因此在一个工作上有两个任务将导致应用程序失败。

我试图分配我的工人资源和任务分配，dask-worker scheduler-ip:port --nprocs 1 --resources process=1但futures = [client.submit(func, f, resources={'process': 1}) for f in futures]没有成功。

我的代码如下：

import dask
from dask.distributed import Client


@dask.delayed
def load():
  ...


@dask.delayed
def foo():
  ...


@dask.delayed
def save():
  ...

client = Client(scheduler-ip:port)

# Process file from a given path
paths = ['list', 'of', 'path']

results = []
for path in paths:
  img = load(path)

  for _ in range(n):
    img = foo(img)

  results.append(save(output-filename))

client.scatter(results)
futures = client.compute(results)

def identity(x):
  return x
client.scatter(futures)
futures = [client.submit(same, f, resources={'process': 1}) for f in futures]

client.gather(futures)

截至目前，我有两种情况：

1-我运行所有输入并且应用程序以MemoryError

2-我运行一个子样本，但它运行如下：

加载(img-1)->加载(img-2)->foo(img-1)->加载(img-3)->...->保存(img-1)->保存(img-2 )->...

TLDR：这就是我想对每个工人做的事情：

加载(img-1)->foo(img-1)->保存(img-1)->加载(img-7)->...

4

1 回答 1

0

这里最简单的事情可能是只用一个线程启动你的工作人员

dask-worker ... --nthreads 1

然后那个工人一次只会开始一件事

于 2019-04-15T14:00:58.533 回答