dask / dask/distributed

Enhancement Request - Scatter based on worker memory availability

Open
#5,451 1 comment 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

Hello, I'm wondering if it would be possible to have a flag for scatter that selects the worker with the least amount of resident memory, and pushes the data to that worker.

Right now scatter does round robin assignment based on the number of cores a worker has-- so if you call scatter in a loop and have 8-core workers, scatter will push data to the same worker 8 times before moving to a different worker. If you are pushing large data arrays such as multi-gigabyte arrow tables, it is very easy to exceed memory usage on an individual worker and crash the cluster.

I've tried to hack together an ad hoc way of bypassing the default round robin behavior:

``` python
scatterlist = []
tasklist = []
# Get worker names to override default scheduler assignment
workers = client.cluster.scheduler_info.get('workers')
worker = list(workers.keys())
# processList is a list of conditions to check and select subset
for i, chunk in tqdm(enumerate(processList)):
data03.select(data03.chunk == chunk)
# Serialize data to arrow before scatter
tmp = data03.to_arrow_table(selection=True, parallel=True)
# Send data via scatter, capture future
swarmlist.append(client.scatter(tmp, direct=True, workers=worker[i]))
# write_data is a call to write to apache parquet files, using hive ordering on S3
tasklist.append(client.submit(write_data, scatterlist[i]))
# Check and release futures as writes finish
for i, future in enumerate(tasklist):
if future.done():
scatterlist[i].release()
```

This is a pretty error prone way to do this-- the scheduler has much better information on which workers are memory burdened; also, there's no good guarantee that worker names/ip addresses will be stable, as they may change during the run. I could fine tune the above jist to use something like a generator on the workernames in the loop to account for worker dropout, and make the assumption that the rotating workers will generally allow enough time for data to be removed, but this seems like the type of task that is probably better handled by dask itself since it has a complete picture of worker status.

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.