dask / dask/distributed

Client.map() keys as a list breaks Dask queue system

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

Description

**Describe the issue**:
This _could_ not be a bug, but if not then it is unclear to me from the docs why dask behaves like this.
When passing a list of values to client.map() to use as keys, the queueing system seems to break.

In the reproducible example:
- If keys is empty or a constant string and I submit 1000 simple jobs, 282 move to processing (i have 282 cores), the rest are queued.
- If keys are a list (length matching the input iterables), they all instantly move to processing.

**Minimal Complete Verifiable Example**:

```python
from distributed import Client, progress

def simple_task_wait_60s(id: int, logger=None):
from time import sleep
sleep(10)
return 0

if __name__ == '__main__':
n = 15000
iterabs = [str(i) for i in range(1000)]

c = Client(address='tcp://xxx.xxx.xxx.xxx:8786')

data = c.map(simple_task_wait_60s,
iterabs,
pure=False,
# key=iterabs
)
progress(data)
results = c.gather(data)

```
The above produces the following output on the dashboard. Notice the number of jobs queued vs. processing:
![image](https://github.com/dask/distributed/assets/53710727/a9d7318a-441f-409f-b156-511fe2555b55)

Uncommenting the line `key=iterabs` produces the following instead. Notice queued=0:
![image](https://github.com/dask/distributed/assets/53710727/05b064f2-5d3a-49de-a40d-e4ff96ca5dbd)

**Anything else we need to know?**:

**Environment**:

- Dask version: dask=2024.5.2, distributed=2024.5.2
- Python version: 3.10
- Operating System: jobs submitted from macOS 14.5, scheduler and worker running on Ubuntu 22.04.3 LTS
- Install method (conda, pip, source): pip (poetry)

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.