Add Sharding Support

[jstriebel]

Update 2023-11-19:

Sharding is now formalized as a codec for Zarr v3 via the Zarr Enhancement Proposal (ZEP) 2.

A new efficient implementation of sharding is being discussed as part of #1569.

---

TLDR: We want to add sharding to zarr. There are some PRs as starting points for further discussions, as well as issue zarr-developers/zarr-specs#127 to update the spec.

• Sharding Prototype I: implementation as translating Store #876,
• a PR in the clean room implementation zarrita in the spirit of the previous PR, but simplified as much as possible
  Sharding Prototype alimanfoo/zarrita#40, and
• another prototype where the translation logic happens purely in the Array class
  Sharding Prototype II: implementation on Array #947

Please see #877 (comment) for a comparison of the implementation approaches.

---

Currently, zarr maps one array chunk to one storage key, e.g. one file for the DirectoryStore. It would be great to decouple the concept of chunks (e.g. one compressible unit) and storage keys, since the storage might be optimized for larger data per entry and might have an upper limit for the number of entries, such as the file block size and maximum inode number for disk-storage. This does not necessarily fit the access patterns of the data, so chunks might need to be smaller than one storage key.

For those reasons, we would like to add sharding support to zarr. One shard corresponds to one storage key, but can contain multiple chunks:

We, this is scalable minds, will provide a PR for initial sharding support in zarr-python. This work is funded by the CZI through the EOSS program.

We see the following requirements to implement sharding support:

1. Add shard abstraction (multiple chunks in a single file/storage object)
2. Chunks should continue to be compressible (blosc, lz4, zstd, …)
3. Chunks should be stored in a continuous byte stream within the shard file
4. Shard abstraction should be transparent when reading/writing
5. Support arbitrary array sizes (not necessarily chunk/shard aligned)
6. Support arbitrary number of dimensions
7. Possibly store chunks in Morton-order within a shard file

With this issue and an accompanying prototype PR we want to start a discussion about possible implementation approaches. Currently, we see four approaches, which have different pros and cons:

1. Implement shard abstraction as a storage handler wrapper
   • Pro: Can combine logical chunks into single shard keys, which are mapped to chunks of an underlying storage handler. Translation of chunks to shard only needs to happen within the minimal API of the storage handler.
   • Pro: Partial reads and writes to the underlying store are encapsulated in a single place.
   • Con: Sharding should be configured and persisted per array rather than per store.
2. Implement shard abstraction as a compressor wrapper (current chunks = shards which contain subchunks).
   • Pro: The current notion of chunks as a storage-key stays unchanged in the storage layer. They just correspond to what we normally call "shards" then.
   • Con: Partial reads and writes per storage-key need to be passed through all intermediate layers in the array implementation.
   • Con: Addressing a sub-chunk is not intended. This would need to change the concept that a chunk is usually read or written as a whole. This is already partially the case for blosc compression, but still asserts that compression handles data of the size of a chunk, even if not all data is needed. This approach would break this assertion and needs significant changes in the data retrieval of the array implementation.
3. Implement shard abstraction via a translation layer on the array
   • Pro: Fits the conceptual level where sharding is configured and persisted per array.
   • Con: Adding sharding logic to the array make the array implementation even more complex. Chunk/Shard-translation and partial reads/writes would need to be added at multiple points.
4. Implement blosc2 into Zarr (special case of 2)
   • Pro: Sharding is already implemented in the compression.
   • Con: Still, partial read and writes need to happen efficiently throughout the array implementation, see cons of approach 2.
   • Con: One core feature of zarr is the decoupling of the manipulation API, storage layer and compression. This approach tightly couples a single compression with sharding, which rather concerns storage handling. It is not possible to use sharding with other compressions then.

Based on this assessment, we currently favor a combination of approach 1 and 3. This keeps the pros of implementing sharding as a wrapper to the storage handler, which is a nice abstraction level. However, sharding should be configured and persisted per array rather than with the store config, so we propose to use a ShardingStore only internally, whereas user-facing configuration happens on the array, e.g. similar to the shape.

To make further discussions more concrete and tangible, I added an initial prototype for this implementation approach: #876. This prototype still lacks a number of features which are noted in the PR itself, but contains clear paths to tackle those.

We invite all interested parties to discuss our initial proposals and assessment, as well as the initial draft PR. Based on the discussion and design we will implement the sharding support, as well as add automated tests and documentation for the new feature.

[d-v-b]

This is great! Can you give (or direct me to) a high-level description of how the scalable minds sharded format differs from the neuroglancer precomputed format? My lay understanding is that the scalable minds format uses a cartesian grid of shards each containing morton-coded chunks, whereas the neuroglancer scheme entails morton coding the chunks first, then partitioning the morton curve-traversed-chunks into shard files. Is this accurate? What are the pros / cons of these two approaches?

cc @jbms

[jbms]

The neuroglancer precomputed format is rather hacky.

• It is desirable to be able to choose the shard size to be any multiple of the base chunk size, but the neuroglancer precomputed format does not allow that due to the way shards are determined.
• In general if your underlying storage system supports key range queries (e.g. S3, GCS, but not a regular filesystem) then it is desirable for your keys to be ordered in a way that matches access patterns. That often means morton or similar order, though it would ideally be configurable to accommodate different intended access patterns. I think in principle the actual format of the storage keys can be considered out-of-scope for the zarr spec itself, as users can define their own mapping implementation that translates the key. But in practice, as we saw with dimension_separator, it would be better for the zarr metadata itself to fully specify how the keys are formatted.

[jakirkham]

cc @FrancescAlted @joshmoore

[jbms]

cc @laramiel

[shoyer]

I think sharding is could make a lot of sense, but it definitely needs to be exposed in the array API. Otherwise, one might inadvertently attempt to simultaneously write different chunks in the same shard which could lead to data corruption errors.

A related concept, which I believe Neuroglancer also supports, is using hashes to group together shards into different directories:

• Sharding array chunks across hashed sub-directories zarr-specs#115

Together with both of these tricks I think we could achieve good scalability even to PB scale arrays.

[FrancescAlted]

We see the following requirements to implement sharding support:

1. Add shard abstraction (multiple chunks in a single file/storage object)
2. Chunks should continue to be compressible (blosc, lz4, zstd, …)
3. Chunks should be stored in a continuous byte stream within the shard file
4. Shard abstraction should be transparent when reading/writing
5. Support arbitrary array sizes (not necessarily chunk/shard aligned)
6. Support arbitrary number of dimensions
7. Possibly store chunks in Morton-order within a shard file

Good to see that you are tackling this. It is also a shame that you don't find the adoption of blosc2/caterva appealing because IMO it currently fulfills most of the requirements above. But I understand that you have your own roadmap and it is fine if you prefer going with a different solution :-)

[joshmoore]

Hi @FrancescAlted. We're not nearly far enough along to say that blosc2/caterva aren't appealing! 😄 (see approach #4: "Implement blosc2 into Zarr") What we realized after chatting with you was that we really didn't know enough to be able to make any of the critical decisions. The goal here is to get a prototype (or straw man if you will) in place so that as a community we can openly discuss the trade-offs, and critically find the right abstraction level. (Is it a Store? a Backend? an Array? Some new layer or mechanism that doesn't exist yet? ... Personally I agree with @jbms that we're going to want a metadata representation for cross-language handling, etc.)

Kudos to @jstriebel for kicking this off and very glad to see people jumping in both here and on #876. Please do consider this the start of a conversation though. Happy to hear more!

~Josh

[rabernat]

It is also a shame that you don't find the adoption of blosc2/caterva appealing because IMO it currently fulfills most of the requirements above

This is absolutely not true Francesc. Speaking as a Zarr core dev, my preferred solution would be to implement sharding with Blosc2. (Not sure that Caterva is needed.) I have been consistent about this position for a while (see #713). I explicitly requested we include evaluation of Blosc2 / Caterva as part of this work.

[FrancescAlted]

This is absolutely not true Francesc. Speaking as a Zarr core dev, my preferred solution would be to implement sharding with Blosc2. (Not sure that Caterva is needed.) I have been consistent about this position for a while (see #713). I explicitly requested we include evaluation of Blosc2 / Caterva as part of this work.

Ok, happy to hear that. It is just that after having a cursory look at the proposal, it seemed to me that "we currently favor a combination of approach 1 and 3" was leaving Blosc2/Caterva a little bit out of the game. But seriously, I don't want to mess up with your decisions; feel free to take any reasonable approach you find better for your requirements.

[jbms]

If blosc2/caterva were used, is there a way to make it compatible with the existing numcodecs, so that for example you could use imagecodecs_jpeg2k to encode individual chunks within the shard?

As far as using caterva as the shard format, it sounds like there would then be 3 levels of chunking: shard, chunk, block. But writing would have to happen at the granularity of shards, and reading could happen at the granularity of blocks, so it is not clear to me what purpose the intermediate "chunk" serves.

Another question regarding blosc2/caterva: what sort of I/O interface does it provide for the encoded data? In general caterva would need to be able to request particular byte ranges from the underlying store ---- if you have to supply it with the entire shard then that defeats the purpose.

[FrancescAlted]

It has been a long week and you are asking a lot of questions, but let me try to do my best in answering (at the risk of providing some nebulous thoughts).

If blosc2/caterva were used, is there a way to make it compatible with the existing numcodecs, so that for example you could use imagecodecs_jpeg2k to encode individual chunks within the shard?

Sorry, but I don't have experience with numcodecs (in fact, I was not aware that jpeg2k was among the supported codecs). I can only say that Blosc2 supports the concept of plugins for codecs and filters; you can have a better sense of how this work at my recent talk at PyData Global (recording here) and our blog post on this topic. I suppose jpeg2k could be integrated as a plugin for Blosc2, but whether or not numcodecs would be able to retrieve existing imagecodecs_jpeg2k is beyond my knowledge.

As far as using caterva as the shard format, it sounds like there would then be 3 levels of chunking: shard, chunk, block. But writing would have to happen at the granularity of shards, and reading could happen at the granularity of blocks, so it is not clear to me what purpose the intermediate "chunk" serves.

That's a valid point. Actually, Blosc2 is already using chunks as you are planning to use shards, whereas our blocks is what you are intending to use as chunks, so currently Blosc2 only has two levels of chunking. I was thinking more along the lines that Zarr could implement shards as Blosc2 frames (see my talk for a more visual description of what a frame is). Indeed you would end with 3 partition levels, but perhaps this could be interesting in some situations (e.g. for getting good slice performance in different dimensions simultaneously even inside single shards).

Another question regarding blosc2/caterva: what sort of I/O interface does it provide for the encoded data? In general caterva would need to be able to request particular byte ranges from the underlying store ---- if you have to supply it with the entire shard then that defeats the purpose.

Blosc2 support plugins for I/O too. Admittedly, this is still a bit preliminary and not as well documented as the plugins for filters and codecs but still, there is a test that may shed some light on how to use them: https://github.com/Blosc/c-blosc2/blob/main/tests/test_udio.c. You may want to experiment with this machinery, but as we did not have resources enough for finishing this properly there might be some loose ends here (patches are welcome indeed).

[jstriebel]

Thanks everyone for lively discussion! I'd like to emphasize what Josh wrote, please do consider this the start of a conversation. We at scalable minds are interested in implementing sharding support in zarr, not in a specific strategy. The initial description simply is our own and surely incomplete assessment so far.

I'll try to answer the different threads about the general approach in this issue:

---

@d-v-b

Can you give (or direct me to) a high-level description of how the scalable minds sharded format differs from the neuroglancer precomputed format?

So far, we mostly use webknossos-wrap (wkw), which uses compressible chunks (called blocks in wkw). As you said, a (cartesian) cube of chunks is grouped into one file, where the chunks are concatenated in Morton-order. Each file starts with some metadata: General information such as the chunk- and shard-sizes (similar to .zarray), and a jumptable to the chunk-wise start-positions iff chunks are compressed. Our access patterns mostly are shard-wise writes and chunk-wise reads.

---

@shoyer Thanks for the hint about zarr-developers/zarr-specs#115, this definitely looks interesting. As I understand it this would be an orthogonal concept, right? The main benefit in sharding as proposed here is that it leverages the underlying grid (e.g. extending a dataset somewhere only touches the surface-shards). One could still design the hashing-function to effectively be cube-based sharding, but I think that this rather "hacks" the idea of hashing and might be more useful as a separate concept. Also the order of chunks within a shard/hash-location can be optimized based on the geometrical grid when using sharding, for hashing this again would be rather hacky. Does this make sense?

---

@FrancescAlted @rabernat blosc & caterva
As far as I understand blosc2, it does itself provide chunked access within a frame, which represents one file. Those chunks can be compressed and filtered. I see two possible concepts, how blosc2 could be integrated:

• a As another compression, which is highly configurable. There might be optimized access for subparts of a file, similar to the current PartialReadBuffer, but this does not replace sharding itself, which is handled on another level, similar to caterva. It might also be possible to only use a blosc2-chunk instead of a frame as a zarr chunk, which seems like the correct abstraction level for a compression.
• b As a major part of zarr, where the compression and filters of blosc2 are used instead of the current compression and filter mechanisms, and blosc2-chunks would serve as chunks in a shard. The current array representation and access must completely be changed.

To me one major appeal of zarr is the separation of storage, array access (e.g. slicing) and compression. Using blosc2 as a sharding strategy would directly couple array access and compression, since slicing must know about the details of blosc2 and use optimized read and write routines. This already is the case for blosc with the PartialReadBuffer and would need to go much further, which I think degrades readability and extensibility of the code. Also, it allows sharding only with blosc2 and compatible compressions & filters. Any custom compressions/filters would first need to be integrated with blosc2 before they could be sharded, which directly couples storage considerations (sharding) with the compression. Implementing sharding within zarr itself would allow that those parts stay decoupled as they are and seems a quite doable addition.

Based on those arguments, I personally would rather tend to integrate blosc2 as another compression (a). This wouldn't be part of this issue here, but definitely a useful addition.

Regarding caterva, I understand that it handles chunking (caterva blocks) and sharding (caterva chunks), but stores this information in a binary header, and is coupled to blosc2:

To me this seems rather like an alternative to zarr, but surely could also be used as another compression (which seemed to be the initial strategy in #713). As already mentioned by @jbms, both caterva chunks and blosc2 frames both allow chunk-wise reads in a sub-part of a file and serve a similar use-case. This seems doubled, but helps to separate the APIs, where caterva cares about data access and blosc2 about compression. This is similar to the question in a, if a zarr-chunk should correspond to a blosc2-frame or a blosc2-chunk.

As you can see, we definitely do consider using blosc2 & caterva for sharding. Atm it simply does not seem to be the most effective way to me. I hope those arguments make sense, please correct me where I'm wrong, and I'm very happy to hear more arguments about using blosc2 or caterva for sharding.

---

PS: Some discussion about our initial implementation draft is also going on in the PR #876. Before investing much more time into the PR, I'd like to reach consensus about the general approach here.

[FrancescAlted]

@jstriebel Yeah, as said, it is up to you to decide whether Blosc2/Caterva can bring advantages to you or not. It is just that I was thinking that this could have been a good opportunity to use Blosc2 frames as a way to interchange data among different packages with no copies. Frames are very capable beasts (63-bit capacity, metalayers, can be single-file or multi-file, can live either on-disk or in-memory), and I was kind of envisioning them as a way to easily share (multidim) data among parties.