dask / dask/distributed

P2P shuffle is slow with string dtypes

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

Description

```python
import coiled
import dask.dataframe as dd
from dask.distributed import wait

cluster = coiled.Cluster(
n_workers=30,
worker_cpu=4,
region="us-east-2", # start workers close to data to minimize costs
arm=True,
)

client = cluster.get_client()
```

```python
# this takes 1m21s
df = dd.read_parquet("s3://coiled-datasets/uber-lyft-tlc/")
df = df.set_index("request_datetime", shuffle="tasks").persist()
_ = wait(df)
```

```python
# this takes 2m12s
df = dd.read_parquet("s3://coiled-datasets/uber-lyft-tlc/")
df = df.set_index("request_datetime", shuffle="p2p").persist()
_ = wait(df)
```

GIL contention is very high during the p2p shuffle (also during tasks) and cpu usage is at 100+%, implying, maybe, that the creation/deletion of lots of Python objects is slowing us down considerably.

cc @hendrikmakait @jrbourbeau

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.