dask / dask/distributed

Resilience for workflows that end in `to_<storage>`

Open
#8,182 6 comments 0 reactions 0 assignees View on GitHub
enhancement
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

# Use case
An embarrassingly parallel workflow, where the number of independent pipelines >> the number of workers, ends with a call to `to_zarr` or some other of form of permanent storage. At some point during the computation, a single worker - let's say one out of 16- crashes unexpectedly.

### Expected behaviour
You lose the pipelines that were halfway through on that worker. They are rescheduled automatically.

### Actual behaviour
You lose 1/16th of *all your work so far*. The scheduler wastes time recomputing data that has already been safely saved to your permanent storage.

### What happened
`to_` calls generate one task per chunk/partition, each emitting `None` as output, which is stored on the workers until the whole computation is complete. The scheduler doesn't do anything special about them - which means that if the worker holding them crashes the scheduler will recompute the whole pipeline leading to them.

In pseudocode:
```python
a = dask.array.zeros(...)
a = a.map_blocks(expensive_processing)
z = a.map_blocks(to_storage, fname="s3://somewhere")
z = z.persist()
client.wait(z)
client.gather(z)
```

# Proposed solution 1: AMM AutoReplicate
Implement a new Active Memory Manager policy, `AutoReplicate`, which automatically creates e.g. 4 replicas for tasks whose sizeof is < 1 kiB. Exclude tasks that have between 1 and 4 dependants.

- XREF https://github.com/dask/distributed/issues/6578

# Proposed solution 2: rewrite graph
If the sizeof of the output of a task is e.g. < 500 bytes AND its pickled size is also < 500 bytes, then the `TaskFinishedMsg` will contain the actual pickled value of the task. The scheduler will then immediately
- change the `run_spec` to `lambda: `
- release all the task's non-shared dependencies

If the worker later dies, the scheduler will just reschedule a trivial `lambda: `
You would see the total number of tasks on the scheduler decrease progressively as the computation progresses.
It also means that large amounts of workers could die at the same time without losing work.

CC @fjetter @hendrikmakait

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.