如何让 dask 将任务调度到不同的 specialized workers？

2026-11-04排障赵师傅

结论

先调低 scheduler 的 distributed.scheduler.worker-saturation。如果每个 specialized worker 只有一个可用的 foo 槽位，可以从 1.0 开始：

distributed:
  scheduler:
    worker-saturation: 1.0

这样，scheduler 会更早将已有任务的 worker 判定为 saturated，并尝试选择其他同样提供 foo=1 的空闲 worker。增大这个值不会提高并行度。值越大，scheduler 越倾向于继续在同一个 worker 上排队，data locality 对调度结果的影响也越明显。

active-task-set-scale 控制 scheduler 同时维护的 active tasks 数量。它会影响 scheduler 能否看到足够多的可运行任务，但不会直接决定任务分配给哪个 worker。建议先调整 worker-saturation，不要同时改动这两个参数。

resource 为什么没有自动均匀分配任务

Dask resource 只用于限制容量，并不负责负载均衡。

假设 worker 声明：

foo=1

提交任务时指定：

resources={"foo": 1}

这表示：

任务只能在提供 foo 的 worker 上运行；
一个 worker 同时最多运行一个消耗 foo=1 的任务；
scheduler 不保证按 round-robin 分配任务；
也不保证每个符合条件的 worker 都能收到任务。

一组 batch tasks 共享同一个 dependency 时，这份 dependency 往往已经位于某个 worker。为了避免传输数据，scheduler 可能继续将同组任务分给这个 worker。后续任务会在该 worker 上排队，等待 foo 释放。即使其他 worker 还有空闲的 foo=1，也未必会被选中。

因此，可能出现“一组 A 在一个 worker 上串行运行，另一组 B 在另一个 worker 上串行运行”的情况。

`worker-saturation` 的作用

worker-saturation 决定 scheduler 可以预先向一个 worker 分配多少工作。scheduler 通常会结合 worker 的线程数和预计执行时间，判断该 worker 是否已经 saturated。

调高这个值后，一个 worker 可以接收更多排队任务，更容易利用 data locality，但其他 worker 可能因此空闲。调低后，scheduler 会更早认为当前 worker 已满，主动将任务发送给其他 worker。设置为 1.0 通常适合需要优先使用空闲 worker、减少本地排队的场景。

对于单线程 worker，可以近似理解为：

允许预分配的任务数 ≈ ceil(worker-saturation × nthreads)

这只是便于理解的近似表达。实际判断还会受到任务时长估算、依赖位置和 Dask 版本实现的影响。

如果一个 worker 有多个线程，却只有一个 foo 槽位，那么 worker-saturation: 1.0 仍可能让它接收多个任务。此时可以继续调低，例如：

distributed:
  scheduler:
    worker-saturation: 0.5

也可以让这类 specialized worker 使用 --nthreads 1，使线程容量与 foo=1 的实际并发能力一致：

dask worker tcp://scheduler:8786 \
  --nthreads 1 \
  --resources "foo=1"

`active-task-set-scale` 的作用

active-task-set-scale 限制 scheduler 在 active task set 中同时保留的任务规模，主要用于控制大型 graph 的调度开销和 scheduler 内存占用。

减小该值后，scheduler 同时考虑的 tasks 会变少。增大该值可以让更多 ready tasks 同时进入活跃集合，但不会覆盖 data locality，也不会强制 scheduler 将任务分散到不同 worker。值太小可能导致部分空闲 worker 暂时收不到工作，值太大则可能增加 scheduler 的 CPU 和内存压力。

只有当 dashboard 显示大量满足依赖条件的任务仍未进入可调度状态，同时 specialized workers 处于空闲状态时，才需要考虑增大这个值。如果任务已经处于 processing 或 queued 状态，只是集中在少数 worker 上，应调整 worker-saturation。

不同 Dask 版本可能没有这个配置，或者其实现细节和默认值有所变化。不要直接套用网上给出的固定默认值，应检查实际运行环境。

配置方法

使用 YAML

将配置写入 scheduler 可以读取的 Dask 配置文件：

distributed:
  scheduler:
    worker-saturation: 1.0

scheduler 需要在读取该配置后启动。只在提交任务的 client 进程中修改配置，未必会影响已经运行的远程 scheduler。

使用环境变量

使用容器部署时，可以设置：

DASK_DISTRIBUTED__SCHEDULER__WORKER_SATURATION=1.0

需要在 scheduler container 上配置该环境变量，然后重启 scheduler。

如果 scheduler 和 workers 共用同一个容器模板，也可以统一注入这个变量，实际读取该参数的仍是 scheduler。

在 Python 中启动本地集群

如果当前 Python 进程负责创建 scheduler，可以在创建集群前设置：

import dask
from distributed import Client, LocalCluster

with dask.config.set({
    "distributed.scheduler.worker-saturation": 1.0,
}):
    cluster = LocalCluster()
    client = Client(cluster)

如果外部 scheduler 已经在运行，这段 client 端配置通常不会动态修改其现有配置。

保持一次性提交所有任务

现有 flow 的提交方式无需修改，batch tasks 仍可一次性提交：

from distributed import wait

futures = []

for group_name, batches in batch_groups.items():
    for batch_id, batch in enumerate(batches):
        future = client.submit(
            process_batch,
            batch,
            shared_dependency,
            resources={"foo": 1},
            key=f"{group_name}-batch-{batch_id}",
            pure=False,
        )
        futures.append(future)

wait(futures)

这里使用唯一的 key，并设置 pure=False，可以防止 Dask 将内容相同的调用作为同一个确定性任务复用。如果每个 batch 的参数本来就不同，并且函数是纯函数，则不一定需要 pure=False。

降低 worker-saturation 不会让一次性提交丢失完整 graph。该参数只影响 scheduler 如何将 ready tasks 分配给 workers。

检查 worker 配置是否符合预期

先确认 scheduler 识别到的四个 worker 都正确声明了资源：

info = client.scheduler_info()

for address, worker in info["workers"].items():
    print(
        address,
        "nthreads=", worker["nthreads"],
        "resources=", worker["resources"],
    )

预期结果类似：

tcp://worker-1:1234 nthreads=1 resources={'foo': 1}
tcp://worker-2:1234 nthreads=1 resources={'foo': 1}
tcp://worker-3:1234 nthreads=1 resources={'foo': 1}
tcp://worker-4:1234 nthreads=1 resources={'foo': 1}

如果只有两个 worker 向 scheduler 注册了 foo=1，或者四个 container 实际连接成了两个多线程 worker，那么修改 scheduler 参数也无法得到四路并行。

还要确认所有 tasks 使用完全一致的 resource 名称。Dask 的 resource 名称区分大小写：

resources={"foo": 1}

和：

resources={"Foo": 1}

表示两个不同的 resource。

如何确定当前默认值

默认值可能随 Dask 版本变化，应在 scheduler 使用的环境中查询：

import dask
import distributed

print("dask:", dask.__version__)
print("distributed:", distributed.__version__)

print(
    "worker-saturation:",
    dask.config.get(
        "distributed.scheduler.worker-saturation",
        default=None,
    ),
)

print(
    "active-task-set-scale:",
    dask.config.get(
        "distributed.scheduler.active-task-set-scale",
        default=None,
    ),
)

如果第二项返回 None，可能是当前版本没有这个配置，也可能是该配置没有显式设置。此时应查看当前版本安装包中的配置说明，不要直接复制其他版本的参数。