TopolearnWhen a large job stalls or behaves oddly, you need a mental model of how the scheduler places tasks and moves data.
A distributed cluster has three roles. The client in your Python session builds graphs and sends them over. The scheduler is a single process that tracks every task's state and decides where it runs. Workers execute tasks in thread pools, store results in memory, and fetch dependencies directly from each other. A nanny process supervises each worker and restarts it if it dies or exceeds memory limits.
Each task moves through states on the scheduler: released, waiting (dependencies not ready), processing (assigned to a worker), memory (result held by a worker), or erred. Workers keep their own matching state machine.
Placement decisions:
- Data locality: a task usually goes to the worker that already holds most of its input bytes, estimating transfer cost from data sizes and bandwidth.
- Queuing: root tasks such as file reads are queued on the scheduler and released as workers free up, which keeps memory steady.
- Work stealing: idle workers take queued tasks from busy ones when the compute saved exceeds the transfer cost.
Resilience: if a worker dies, the scheduler recomputes lost results from the graph's lineage. A task that keeps killing workers is marked as erred after allowed-failures attempts (default 3), with a KilledWorker error. The usual cause is a task that uses too much memory.
The scheduler is single-threaded Python, so very high task rates, roughly thousands per second, make it the bottleneck. That's another reason to keep tasks coarse.
Example
from dask.distributed import Client
if __name__ == "__main__":
with Client(n_workers=2, threads_per_worker=1, processes=False) as client:
a = client.submit(lambda: list(range(100_000)), key="big-data")
b = client.submit(len, a, key="needs-big") # goes where 'big-data' lives
print(b.result()) # 100000
who_has = client.who_has([a, b])
print(who_has["big-data"] == who_has["needs-big"]) # True (data locality)
info = client.scheduler_info()
print(len(info["workers"])) # 2
states = client.run_on_scheduler(
lambda dask_scheduler: dask_scheduler.tasks["needs-big"].state)
print(states) # memoryRemember
- Client builds graphs; one scheduler tracks task states; workers compute and share data
- Placement favours data locality, with queuing and work stealing to balance load
- Lost results are recomputed; repeated crashes surface as KilledWorker
Watch out for
- Seeing KilledWorker and retrying blindly instead of shrinking the memory used per task.