dask / dask/distributed

P2P Shuffle blows up the task graph with increasing number of partitions

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

Description

**Describe the issue**:

We put the list of partitions into the graph which blows it up to 400mb for the example below. We added a workaround for dask-expr that unpacks the list in shuffle_transfer, but we should fix things properly via a delayed object or a task that returns the list once for all shuffle tasks

**Minimal Complete Verifiable Example**:

```python
import dask
import numpy as np

df = dask.datasets.timeseries(
dtypes={
"a": int,
"b": float,
"c": float,
},
start="1960-01-01",
end="1991-01-01",
freq="200ms",
partition_freq="1D",
).reset_index().shuffle(on="timestamp")
df = df.persist()
```

We patched this in dask-expr for the moment, so main won't exhibit the issue anymore

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

@hendrikmakait and I tried the task solution but this caused a lot of data transfer on the cluster which was unexpected

Screenshot 2024-07-19 at 13 01 07

**Environment**:

- Dask version: 2024.07
- Python version: 3.12
- Operating System: mac
- Install method (conda, pip, source):

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.