Missing spilled key
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
**What happened**:
I was looking to upgrade Dask and Distributed to a more recent version to patch the reported security vulnerability, but I encountered some issues with the following snippet of the example code. The code worked fine (or behaved the same) until version `2021.02`. Returning the whole Pandas series using `df.baz.compute()` started to fail from `2021.03` onward.
```
....
df = df.join(df[df.bar == "a"].set_index("foo"), on="foo", resuffix="_a")
df = df.join(df[df.bar == "b"].set_index("foo"), on="foo", resuffix="_b")
print(df.head()) # printed successfully
print(df.tail()) # printed successfully
print(df.map_partitions(len).compute()) # printed successfully
print(df.baz.unique().compute()) # print successfully
print(df.baz.compute()) # failed
```
In the `2021.03` version, Dask/Distributed appeared to fail in deserializing at `df.baz.compute()`:
```
distributed.protocol.core - CRITICAL - Failed to deserialize
Traceback (most recent call last):
File "/lib/python3.8/site-packages/distributed/protocol/core.py", line 104, in loads
return msgpack.loads(
File "msgpack/_unpacker.pyx", line 194, in msgpack._cmsgpack.unpackb
File "/lib/python3.8/site-packages/distributed/protocol/core.py", line 96, in _decode_default
return merge_and_deserialize(
File "/lib/python3.8/site-packages/distributed/protocol/serialize.py", line 454, in merge_and_deserialize
return deserialize(header, merged_frames, deserializers=deserializers)
File "lib/python3.8/site-packages/distributed/protocol/serialize.py", line 388, in deserialize
return loads(header, frames)
File "/lib/python3.8/site-packages/distributed/protocol/serialize.py", line 75, in pickle_loads
if writeable[i] == mv.readonly:
IndexError: tuple index out of range
*** IndexError: tuple index out of range
```
In the `2021.04` and `2021.05` versions, Dask/Distributed raised `pynvml.NVMLError_Uninitialized: Uninitialized` error, but it looked to be resolved in the more recent versions. Therefore, I didn't pay attention to these two versions.
Whereas in the version `2021.06` to `2021.12`, Dask/Distributed appeared to lose track of a spilled key at `df.baz.compute()`:
```
distributed.worker - ERROR - Exception during execution of task ('blocks-475b393a0e9dc7abcb0387087f782888', 0).
Traceback (most recent call last):
File "/lib/python3.8/site-packages/distributed/worker.py", line 3185, in execute
args2, kwargs2 = self._prepare_args_for_execution(ts, args, kwargs)
File "lib/python3.8/site-packages/distributed/worker.py", line 3307, in _prepare_args_for_execution
data[k] = Actor(type(self.actors[k]), self.address, k, self)
KeyError: "('merge-4e80b8642726e77a65d0f6036681eb12', 0)"
distributed.worker - ERROR - Exception during execution of task ('len-03a53e93215741edfdf491efc0e3fe1a', 3).
Traceback (most recent call last):
File "/lib/python3.8/site-packages/distributed/worker.py", line 3303, in _prepare_args_for_execution
data[k] = self.data[k]
File "/lib/python3.8/site-packages/zict/buffer.py", line 78, in __getitem__
return self.slow_to_fast(key)
File "/lib/python3.8/site-packages/zict/buffer.py", line 71, in slow_to_fast
cb(key, value)
File "/lib/python3.8/site-packages/distributed/spill.py", line 62, in _on_retrieve
self.spilled_total -= self.spilled_by_key.pop(key)
KeyError: "('merge-4e80b8642726e77a65d0f6036681eb12', 3)"
```
**What you expected to happen**:
I expected them not to fail.
**Minimal Complete Verifiable Example**:
Sorry, I tried to reproduce an MCV, but no luck.
**Anything else we need to know?**:
In the version `2021.06` to `2021.12`, I also noticed the (spilled) key was missing in 1 - 2 partitions. It was not deterministic.
I could also reproduce the issue in an Ubuntu container.
**Environment**:
- Dask version: 2021.03 and above
- Python version: Python3.8
- Operating System: macOS
- Install method (conda, pip, source): poetry
Cluster Dump State:
Contributor guide
Assessment
This issue has not been assessed yet.