dask / dask/distributed

Incorrect documentation about `workers` keyword in Data Locality article

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

Description

The documentation in the [_Specify Workers in Compute/Persist_ section in the Data Locality article](http://distributed.dask.org/en/stable/locality.html#specify-workers-with-compute-persist) seems to indicate that the `workers` keyword in `Client.compute()` accepts a dictionary.

**What happened**:

In practice, the `workers` keyword does not accept a dictionary.

**What you expected to happen**:

Expected that the `workers` keyword would accept the dictionary AND the tasks would execute on the workers specified by the dictionary.

**Minimal Complete Verifiable Example**:

```python
import dask
dask.__version__
```

OUT:
'2021.11.2'

```python
from dask.distributed import Client

client = Client()
```

```python
worker_addresses = [x for x in client.cluster.scheduler_info['workers']]
worker_addresses
```

OUT:
['tcp://127.0.0.1:35297',
'tcp://127.0.0.1:36365',
'tcp://127.0.0.1:41759',
'tcp://127.0.0.1:38571']

```python
from time import sleep
def add(x, y):
sleep(1)
return x + y

def subtract(x, y):
sleep(1)
return x - y

def multiply(x, y):
sleep(2)
return x * y
```

```python
from dask import delayed
x = 12
y = 8
add_result = delayed(add)(x, y)
subtract_result = delayed(subtract)(x, y)
multiply_result = delayed(multiply)(add_result, subtract_result)

multiply_result.visualize()
```

OUT:
![image](https://user-images.githubusercontent.com/46224135/147493704-bfab62f3-cd3b-40f5-872c-d160206180dd.png)

```python
futures = client.compute(multiply_result, workers={multiply_result: worker_addresses[0], add_result: worker_addresses[1]})
futures
```

OUT:

Future: multiply
status:

pending,

type: NoneType,

key: multiply-4c43bba2-8553-4a09-a7af-63d78db6cf05

distributed.protocol.core - CRITICAL - Failed to Serialize
Traceback (most recent call last):
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/protocol/core.py", line 76, in dumps
frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/msgpack/__init__.py", line 35, in packb
return Packer(**kwargs).pack(o)
File "msgpack/_packer.pyx", line 292, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 298, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 295, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 264, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 264, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 264, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 229, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 289, in msgpack._cmsgpack.Packer._pack
TypeError: can not serialize 'Delayed' object
distributed.comm.utils - ERROR - can not serialize 'Delayed' object
Traceback (most recent call last):
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/comm/utils.py", line 33, in _to_frames
return list(protocol.dumps(msg, **kwargs))
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/protocol/core.py", line 76, in dumps
frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/msgpack/__init__.py", line 35, in packb
return Packer(**kwargs).pack(o)
File "msgpack/_packer.pyx", line 292, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 298, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 295, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 264, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 264, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 264, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 229, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 289, in msgpack._cmsgpack.Packer._pack
TypeError: can not serialize 'Delayed' object
distributed.batched - ERROR - Error in batched write
Traceback (most recent call last):
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/batched.py", line 93, in _background_send
nbytes = yield self.comm.write(
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/tornado/gen.py", line 762, in run
value = future.result()
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/comm/tcp.py", line 250, in write
frames = await to_frames(
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/comm/utils.py", line 50, in to_frames
return _to_frames()
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/comm/utils.py", line 33, in _to_frames
return list(protocol.dumps(msg, **kwargs))
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/protocol/core.py", line 76, in dumps
frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/msgpack/__init__.py", line 35, in packb
return Packer(**kwargs).pack(o)
File "msgpack/_packer.pyx", line 292, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 298, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 295, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 264, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 264, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 231, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 264, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 229, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 289, in msgpack._cmsgpack.Packer._pack
TypeError: can not serialize 'Delayed' object

```python
futures = client.compute(multiply_result, workers={'multiply_result': worker_addresses[0], 'add_result': worker_addresses[1]})
futures
```
OUT:

Future: multiply
status:

cancelled,

type: NoneType,

key: multiply-4c43bba2-8553-4a09-a7af-63d78db6cf05

distributed.core - ERROR - unhashable type: 'dict'
Traceback (most recent call last):
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/core.py", line 596, in handle_stream
handler(**merge(extra, msg))
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 4528, in update_graph_hlg
return self.update_graph(
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 4765, in update_graph
w = self.coerce_address(w)
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 7353, in coerce_address
if addr in parent._aliases:
TypeError: unhashable type: 'dict'
distributed.core - ERROR - Exception while handling op register-client
Traceback (most recent call last):
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/core.py", line 530, in handle_comm
result = await result
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 5359, in add_client
await self.handle_stream(comm=comm, extra={"client": client})
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/core.py", line 596, in handle_stream
handler(**merge(extra, msg))
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 4528, in update_graph_hlg
return self.update_graph(
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 4765, in update_graph
w = self.coerce_address(w)
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 7353, in coerce_address
if addr in parent._aliases:
TypeError: unhashable type: 'dict'
tornado.application - ERROR - Exception in callback functools.partial(. at 0x7f49d0c5a790>, exception=TypeError("unhashable type: 'dict'")>)
Traceback (most recent call last):
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/tornado/ioloop.py", line 741, in _run_callback
ret = callback()
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/tornado/tcpserver.py", line 331, in
gen.convert_yielded(future), lambda f: f.result()
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/comm/tcp.py", line 519, in _handle_stream
await self.comm_handler(comm)
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/core.py", line 530, in handle_comm
result = await result
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 5359, in add_client
await self.handle_stream(comm=comm, extra={"client": client})
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/core.py", line 596, in handle_stream
handler(**merge(extra, msg))
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 4528, in update_graph_hlg
return self.update_graph(
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 4765, in update_graph
w = self.coerce_address(w)
File "/home/rehan/notebook_testing/jupyter-env/lib/python3.8/site-packages/distributed/scheduler.py", line 7353, in coerce_address
if addr in parent._aliases:
TypeError: unhashable type: 'dict'

```python
futures = client.compute(multiply_result, workers=[worker_addresses[0]])
```

```python
futures
```

OUT:

Future: multiply
status:

pending,

type: NoneType,

key: multiply-4c43bba2-8553-4a09-a7af-63d78db6cf05

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

Apart from fixing the documentation, **is there an alternative way of calling the compute method and specifying that the intermediate tasks be executed on a specific worker?**

**Environment**:

- Dask version: 2021.11.2
- Python version: 3.8.10
- Operating System: Linux (ubuntu 20.04)
- Install method (conda, pip, source): pip

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.