Data loss during p2p shuffle
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
**Describe the issue**:
After performing actions that involve (p2p?) shuffling, e.g., shuffling or merging, the number of rows in a data frame _slightly_ (which is different from issue https://github.com/dask/distributed/issues/4386) decreases.
**Minimal Complete Verifiable Example**:
I could not reproduce using a synthetic example, but I have a data frame with a single column "id" of type `int64`. Min value is 23, max value is 301,870,552 (so, no integer overflows or anything like that), no missing values. It is partitioned into 32 partitions.
```python
ddf = dd.read_parquet(url).persist()
print(ddf.shape[0].compute())
# 301870552
for i in range(5):
n_rows = ddf.shuffle(on="id", npartitions=32).shape[0].compute()
print(n_rows)
# 301870552
# 301870552
# 301870544 <-- smaller
# 301870552
# 301870552
for i in range(5):
n_rows = ddf.shuffle(on="id", npartitions=32, shuffle="p2p").shape[0].compute()
print(n_rows)
# 301870552
# 301870552
# 301870552
# 301870552
# 301870545 <-- smaller
for i in range(10):
n_rows = ddf.shuffle(on="id", npartitions=32, shuffle="tasks").shape[0].compute()
print(n_rows)
# 301870552
# ...
# 301870552
```
**Anything else we need to know?**:
I am running it in a k8s cluster with 32 workers with 8Gi of RAM, scattered across 7 nodes.
The problem seem to disappear, when
- I am doing the same as above, but on a single worker. However, even introducing the second one, already creates this issue.
- I use a smaller data frame.
- Overriding automatic shuffle method with "tasks" explicitly seems to help, but since this error appears randomly, there is no guarantee that this is a solution.
Originally discovered while using merging. Merging leads to silent loss of data. Joining using index and `.join()` results in an exception:
> TypeError: cannot do slice indexing on Index with these indexers [1937738] of type int64
although items with ID 1937738 are present in both tables.
**Environment**:
- Dask version: 2023.11.0
- Python version: 3.10
- Operating System: Linux on remote cluster, MacOS on client.
- Install method (conda, pip, source): Docker image (ghcr.io/dask/dask:2023.11.0-py3.10)
Contributor guide
Assessment
This issue has not been assessed yet.