dask not respecting worker resources
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
I believe this issue is related to this [SO](https://stackoverflow.com/questions/56176567/dask-not-respecting-worker-directive-constraint-workers-allow-other-work) post.
I'm finding that, despite resource restrictions, the dask-scheduler will assign keys to worker nodes even when I'm specifying otherwise.
For example:
```
from dask.distributed import LocalCluster, Client
cluster = LocalCluster(n_workers=0)
cluster.start_worker(ncores=1) # resources={'CPU': 1}, no resources to compute
client = Client(cluster.scheduler.address)
fut = client.submit(lambda x=1: x+1, resources={'CPU':1})
client.who_has() # here we expect to see nothing as no workers have resources
fut2 = client.submit(lambda x=1: x+1)
client.who_has() # yep, assigned because it doesn't need resources
import dask.dataframe as dd
import pandas.util.testing as tm
import os
files = dd.from_pandas(tm.makeTimeSeries(1000, freq='10ms'), npartitions=4).to_csv('example*.csv')
# simulate some pipeline where data is read and transformed
fut3 = client.compute(dd.read_csv('example*.csv').to_delayed(), resources={'CPU':1})
client.who_has()
# now we see that the scheduler has placed the delayed keys
# onto the worker even though it has no resources to compute.
# "('from-delayed-238ec9c6404d8e52399becf66971834c', 2)": ('tcp://127.0.0.1:44429',),
# "('pandas_read_text-read-block-from-delayed-238ec9c6404d8e52399becf66971834c', 2)": (),
[os.remove(f) for f in files]
```
Likewise, I'm finding similar behavior with methods like `read_parquet`.
```
dask.__version__
'1.2.2'
```
Contributor guide
Assessment
This issue has not been assessed yet.