dask / dask/distributed

P2P DataFrame performance - Disk overhead

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

Description

P2P is currently writing way too much data to disk

```python
cluster = Cluster(
n_workers=10,
worker_vm_types='m6i.2xlarge',
)
client = cluster.get_client()
df = timeseries(
start="2020-01-01",
end="2035-03-30",
freq="1200ms",
partition_freq="24h",
dtypes={str(i): float for i in range(100)},
)

res = df.shuffle("0").size
res.compute()
```

![image](https://github.com/dask/distributed/assets/8629629/d9eb7a53-4ddc-4374-92ce-9778fac68ae9)

This is about three times more data written to disk than what is transmitted over network. This is caused by us writing the schema + schema metadata to disk for every single micro-shard (I could reduce the overhead by stripping the pandas metadata from the schemas to just 2x and could trace the remaining overhead to the rest of the schema through micro benchmarks).

I believe this was introduced https://github.com/dask/distributed/pull/7410 which removed some logic that would write the schema only to the buffer if the buffer/file was empty, therefore ensuring that we would not write duplicate metadata. This means that right now we're also writing this redundant data to the comm buffer. However, the comms are not impacted as much since comm shards are typically larger than disk shards (comm is split in #Worker shards will disk is split to #OutputPartition shards and typically #OutputPartitions >> #Workers)

cc @hendrikmakait

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.