python - 在 Dask map 功能中设置最大工作人员数

Question

我有一个 Dask 进程可以触发 100 个具有映射功能的工作人员:

worker_args = .... # array with 100 elements with worker parameters 

futures = client.map(function_in_worker, worker_args) 
worker_responses = client.gather(futures)

我使用 docker，其中每个 worker 都是一个容器。我已将 docker 配置为生成 20 个 worker /容器，如下所示:

docker-compose up -d --scale worker=20

问题是我的机器崩溃了，因为 map 函数并行触发了 20 个 worker，这使得内存和 CPU 超过了最大值。

我想保留 20 个 worker 的配置，因为我将这些 worker 用于不需要大量内存的其他功能。

如何将 map 函数限制为 5 个 worker 并行？

Answer 1

dask 不会根据空闲的 worker 数量动态调整 worker 资源。在您提供的示例中，一旦启动了20个worker，如果只使用了5个worker，则不会分配剩余15个空闲worker的资源。

如果这是可以接受的(例如，因为空闲资源正在被外部程序使用)，那么将工作限制为 5 个 worker 的一种方法是通过 workers kwarg 明确指定它们到 。映射调用:

# instantiate workers
from distributed import Client
c = Client(n_workers=20)

# select at most 5 workers from the available list
selected_workers = list(c.scheduler_info()['workers'])[:5]

dummy_function = lambda x: x**2
futs = c.map(dummy_function, range(10), workers=selected_workers)

控制工作负载分配的另一种方法是使用resources kwarg，请参阅以下相关答案:0 , 1 , 2 , 3 .