dask / dask/distributed

Data loss during p2p shuffle

Open
#8,426 22 comments 1 reaction 0 assignees View on GitHub
bug needs info 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

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.