scatter dask collections
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
If a user has a large amount of local numpy or pandas data and wants to use it with dask collections, there isn't a straightforward way, today, to push it from the client to the workers:
- `scatter()` will just return a Future to the numpy data - which dask.array, dask.dataframe, xarray, etc. won't know what to do with.
- Creating a local dask collection and then calling `persist()` will push the whole thing through the scheduler, likely crashing it due to memory pressure.
# Workaround
```python
test = np.zeros(10)
a = da.from_array(test)
dsk = client.scatter(dict(a.dask), broadcast=True)
a = da.Array(dsk, name=a.name, chunks=a.chunks, dtype=a.dtype, meta=a._meta, shape=a.shape)
```
# Proposed design
`distributed.Client.scatter()` could automatically recognize when one or more arguments are dask collections. When it finds some, it could extract their underlying graph, assert that all nodes are data (not callable), scatter the graph values and rebuild the collection automatically - just like `persist()` does.
This should be documented in the high level documentation: "What if I have a huge amount of data on the client?".
The answer should explain that, to begin with, you should try loading data from the workers instead.
Failing that, you should use scatter() instead of persist().
In `persist()`, write a warning when the user tries pushing through a dask graph with hundreds of mb of data, pointing to the documentation.
The snippet above would become:
```python
test = np.zeros(10)
a = da.from_array(test)
a = client.scatter(a)
```
# Alternative design
Make persist() more intelligent, making it switch to scatter() for data chunks beyond a certain threshold.
I find this problematic - scatter() has problems with robustness and flat out won't work on dynamic clusters with 0 workers - so I'd rather keeping the different route explicit.
Contributor guide
Assessment
This issue has not been assessed yet.