dask / dask/distributed

Can we get rid of sharding?

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

Description

The `distributed.comm.shard` setting, which defaults to 64 MiB, is supposed to split a buffer that is larger than the shard size into separate buffers. The intent is to prevent issues that commonly happen beyond 2 GiB with various compression and network protocols.

This setting does nothing for dask.dataframe:

```python
>>> import numpy
>>> import pandas
>>> from distributed.protocol import serialize_bytelist
>>> a = numpy.random.random((2**23,2)) # 128 MiB
>>> frames = serialize_bytelist(a)
>>> [memoryview(f).nbytes for f in frames]
[32, 236, 67108864, 67108864]
>>> frames = serialize_bytelist(pandas.DataFrame(a))
>>> [memoryview(f).nbytes for f in frames]
[40, 123, 554, 134217728, 0]
```

This should be a straightforward bug to fix. However, it also strongly indicates that the whole system tries to solve a purely hypothetical problem. No dask.dataframe user, to my knowledge, has ever complained about crashes.

Do we have an inventory of protocols that break beyond 2 GiB, and/or create temporary deep copies of whole buffers while they work on them?

Sharding causes major performance problems with compression, where any algorithm that doesn't support a `decompress_into` function - _all currently supported algorithms; the only ones I know that do are cramjam and blosc2_ - is plagued by deep-copies and memory spikes because of this:
- #7433

Sharding also adds complexity to the already bloated serialization layer.

CC @milesgranger

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.