ZEP0003: Variable Chunk Sizes

[ivirshup]

Posting ZEP 0003 here for discussion (as part of the ZEP workflow) (@MSanKeys963)

Still very drafty, but could use some discussion. Up to date copy at: https://github.com/zarr-developers/zeps/blob/main/draft/ZEP0003.md

ZEP 3 — Variable chunking

Authors:

• Martin Durant (@martindurant), Anaconda, Inc.
• Isaac Virshup (@ivirshup), Helmholtz Munich

Status: Draft

Type: Specification

Created: 2022-10-17

Abstract

To allow the chunks of a zarr array to be rectangular grid rather than a regular grid, with the chunk lengths along any dimension a list of integers rather than a single chunk size.

Motivation and Scope

Two specific use cases have motivated this, given below. However, this generalisation of Zarr's storage
model can be seen as an optional enhancement, and the same data model as currently used by dask.array.

• when producing a kerchunked dataset, the native chunking of the targets cannot be changed. It is common to have non-regular chunking on at least one dimension, such as a time dimension with one sample per day and chunks of one month or one year. The change would allow these datasets to be read via kerchunk, and/or converted to zarr with equivalent chunking to the original. Such data cannot currently be represented in zarr.
• awkward arrays, ragged arrays and sparse data can be represented as a set of one-dimensional arrays, with an appropriate metadata description convention. The size of a chunks of each component array corresponding to a logical chunk of the overall array will not, in general be equal with each other in a single chunk, nor consistent between chunks, as each row in the matrix can have a variable number of non-zero values
• sensor data, may not come in fixed increments; variably chunked storage would be great for parallel writing. With variable chunk sizes, just need to make sure offsets are correct once done. Otherwise, write locations for chunks are dependent on previous chunks.
• in some cases, parts of the overall data array may have very different data distributions, and it can be very convenient to partition the data by such characteristics to allow, for example, for more efficient encoding schemes.
• when filtering regular table data on one column and applying to other columns, you necessarily end up with an unequal number of values in each chunk, which zarr does not currently handle.

Usage and Impact

Creation

zarr.create(1000, chunks=((100, 300, 500, 100),))

Backward Compatibility

This change is fully backward compatible - all old data will remain usable. However, data written with variable chunks will not be readable by older versions of Zarr. It would be reasonable to wish to backport the feature to v2.

Related Work

Dask

dask.array uses rectangular chunking internally, and is one of the major consumers of zarr data. Much of the code translating logical slices into slices on the individual chunks should be reusable.

Parquet/ Arrow

Arrow describes tables as a collection of record batches. There is no restriction on the size of these batches. This is not only very flexible, but can be used as an indexing strategy for low cardinality columns within parquet.

dataset_name/
  year=2007/
    month=01/
       0.parq
       1.parq
       ...
    month=02/
       0.parq
       1.parq
       ...
    month=03/
    ...
  year=2008/
    month=01/
    ...
  ...

This feature was cited as one of the reasons parquet was chose over zarr for dask dataframes: dask/dask#1599

awkward array

zarr-developers/zarr-specs#62

Implementation

It is to be hoped that much code can be adapted from dask.array, which already allows variable chunk sizes on each dimension.

Alternatives

Just tune chunk sizes

zarr-developers/zarr-specs#62 (comment)

Discussion

References and Footnotes

• Previous discussion:
  • Zarr Dask Table dask/dask#1599
  • Protocol extensions for awkward arrays zarr-developers/zarr-specs#62
  • Handling arrays with non-uniform chunking zarr-developers/zarr-specs#40
  • Chunk spec zarr-developers/zarr-spec#7

Copyright

This document has been placed in the public domain.

[ivirshup]

@martindurant, thinking about construction API:

Can we start with an array of regular chunks, and then append new chunks of different sizes? What's the API here?

Currently array.append will rechunk it's arguments as it adds them to array. Instead, we could allow appending chunks in whatever size they're currently in. We would at least need to add another argument to append to allow this, so we can maintain current behavior.

Potential usage:

a = zarr.ones(5, chunks=3)
a.chunks
# (3,)

Current behavior:

a.append(zarr.ones(3, chunks=2))
a.chunks
# (3,)

Instead, we could have:

a.append(zarr.ones(3, chunks=2), variable_chunks=True)
a.chunks
# ((3, 2, 2, 1),)

^ This gets a little weird due to the misalignment of regular chunk boundaries in each input array.

We could also say this is out of scope for the ZEP, which will only address creation.

[martindurant]

Appending the chunks as they are in the new data makes sense, but it would also be nice to allow specifying chunks. For example, where data is being updated one timepoint at a time, we probably do not want to end up with a bunch of [1] chunk lengths. Also we need to decide what happens at the outer edge of the last chunk, if it not full. Presumably, this process would allow for the conversion of an array which had been regularly chunked to an irregular one.

[ivirshup]

Presumably, this process would allow for the conversion of an array which had been regularly chunked to an irregular one.

To do this, wouldn't we need to either:

• Rewrite the last chunk, so it was variably sized and didn't have fill_values
• Allow having extra fill values at the end of any chunk in a variably chunked array

I think the former would be far less error prone, but I don't love that it requires chunk IO at conversion time.

For example, where data is being updated one timepoint at a time, we probably do not want to end up with a bunch of [1] chunk lengths.

I think I'd be fine to leave this to the user. If someone really wants to do this they could – but the real solution here is for them to buffer their own writes, right?

[normanrz]

Thanks for providing this ZEP. I think support for variable-length chunks would make Zarr more useful for many new applications.

I wonder about the scalability of the proposed design. If you have a large number of chunks (in the variable-sized dimension), they would all need to be listed in the metadata, correct? Upon creation the metadata would need to be synchronized, which could limit parallel writes.

[martindurant]

Correct, the outline here is to store the chunks in the JSON metadata. There are other options, but it's worth pointing out that this amounts to N-1D lists for an N-dimensional array. Example: array of size 100001000010000 (=8TB for int64) with unlikely small chunks of 101010 would have chunks arrays of length

len(json.dumps([[10] * 1000]*3))

=12k (or 9k stripping whitespace), and takes 160us to parse with builtin json.

I don't understand the point about synchronisation, but it may be important. I can imagine three ways to write:

• the chunks sizes are known beforehand (so write the metadata first, as is normal); if the writes match these boundaries, they are fully parallel
• the chunk sizes are gathered from the chunks, and the metadata updated when done; the writes are parallel, but a barrier task must finalise the process.

Both these are things that probably dask would want to get right, but perhaps doesn't change the zarr code at all.

[ivirshup]

Upon creation the metadata would need to be synchronized, which could limit parallel writes.

The way I'm understanding this is the case where you are receiving data in non-fixed batch sizes, and gradually creating an array. So parallel (or at least async) appends. Is this right @normanrz?

Though, I don't think this would necessarily be worse than for regularly chunked arrays since chunk names and array size already have to be synced.

[rabernat]

I think we can keep things simple for now and assume that the the variable chunk sizes are know at array creation time. This mirrors the current zarr behavior, where you have to specify the constant chunk sizes at array creation time. It's up to the application to figure this out before it starts to write.

The main edge case is appending.

[ivirshup]

Another use case from @normanrz: ome/ngff#33 (comment)

We recently implemented the mesh format from Neuroglancer in webKnossos: https://github.com/google/neuroglancer/blob/master/src/neuroglancer/datasource/precomputed/meshes.md

It's been great for our purposes:

• Supports billions of meshes
• Uses draco encoding, which is very storage-efficient
• Supports chunked mesh loading, which is important for very large objects
• Supports multi level-of-detail for progressive loading

I think that format would be a great candidate to be adopted by OME-NGFF.

...

One implementation on top of zarr would be to store each meshfile as on chunk (e.g. in a 2D uint8 array). This would create a lot of chunk files and might create some issues, because the chunks will have different byte lengths.

Looking at the documentation of the format, it overlaps quite a bit with the to-be-written oct tree use case (cc: @kevinyamauchi)

[jbms]

I think it may be helpful to distinguish between the key-value store layer (what the zarr spec calls a store) and the zarr array layer that is provided on top of that. The neuroglancer multiscale mesh format is quite naturally implemented on top of the zarr key-value store layer, but I don't think there is any added value in trying to shoehorn it into the zarr array layer.

Given that ome-ngff is currently organized around zarr group hierarchies I'm not sure what the best way to integrate it might be. Perhaps we should provide a way in zarr v3 for a zarr group hierarchy to contain prefixes/directories with arbitrary non-zarr array data.

[normanrz]

I think it may be helpful to distinguish between the key-value store layer (what the zarr spec calls a store) and the zarr array layer that is provided on top of that. The neuroglancer multiscale mesh format is quite naturally implemented on top of the zarr key-value store layer, but I don't think there is any added value in trying to shoehorn it into the zarr array layer.

It seems to me, in addition to the zarr array layer it would be helpful to have a (zarr) hash map layer.

[jbms]

Hash map in particular, or key-value store generally (which may likely be implemented using an ordered data structure)?

[normanrz]

Fair point. A key-value store, but with multiple key-value pairs per stored object (i.e. sharding).

[jbms]

Sharding, as it has been proposed, is very much tied to the array structure itself, just like chunking, and is built on top of a base key-value store. If we are talking about storing arbitrary data, then I don't think it would be that useful to provide sharding in a generic way, since it would not be able to be tied to the structure of the data. However, the non-array format data might have its own sharding-like scheme. It would also be natural to allow zarr storage transformers that are not specific to arrays to be applied to these non-array prefixes.

[martindurant]

POC: zarr-developers/zarr-python#1483 (it works, ish)

[ivirshup]

I've been working on variable chunk based kerchunk concatenation (fsspec/kerchunk#374) on top of the zarr-python POC. Also works, ish.

However, pretty much requires that data either fits exactly within chunk boundaries or was written with variable length chunks.

[martindurant]

requires that data either fits exactly within chunk boundaries or was written with variable length chunks

Do you mind expanding on what you mean by this?

[ivirshup]

Sure. I would not be able to kerchunk concatenate these arrays:

import zarr, numpy as np
from numcodecs import LZ4

arrays = [
    zarr.array(np.arange(15), chunks=(10,), compression=LZ4()),
    zarr.array(np.arange(15, 30), chunks=([5, 10],), compression=LZ4()),
]

The second chunk of the first array is only half used. Because AFAIK we have no way to indicate we only want the first five values we can't combine these. If the data were uncompressed, we could do this with a range request, but we can't do this with arbitrary compression techniques.

If we could do range requests into the decompressed chunk, then we could maybe do this.

Does that explain this? Or do you see another way around it?

[ivirshup]

In your comment on another issue (fsspec/kerchunk#218 (comment)) you say:

It is certainly correct that kerchunk cannot change the fundamental chunk sizes in the underlying files, and also that zarr requires all the chunks along a given axis to be the same size (except the last). I am hoping to remove this requirement, but that need agreement within the zarr project itself.

I only see how this proposal removes the second requirement "that zarr requires all the chunks along a given axis to be the same size (except the last)". It doesn't solve "that kerchunk cannot change the fundamental chunk sizes in the underlying files", which is what you'd need if you want to slice out any padding.

[martindurant]

Agreed, if there is any extra manipulation that needs to be done to chunks, such as slicing off padding, this could not be achieved by varchunks alone.

Since kerchunk is in a position that it could assume many per-chunk operations (per-chunk scale-offset came up before) since extra information about each chunk could be stored along with the reference. However, this kind of thinking could rapidly end up reimplementing much of numcodecs and zarr itself within kerchunk. Perhaps this is where zarrita's simpler codebase could come into play, with some specific integration points for, er, "flexible" datasets.

[TomNicholas]

Another use case for variable-length chunks, in Cubed. Cubed uses Zarr arrays to store intermediate results between steps of serverless distributed computations. Any computation step that changes the chunks to be irregular (e.g. certain groupby operations) is currently forbidden by Zarr not supporting irregular chunks. (xref cubed-dev/cubed#312)

[d-v-b]

I think variable-length chunks would allow incrementally rechunking an array in-place, which could be very convenient

[martindurant]

I'm not sure how much the spec can do about consistency - the metadata is either in sync with the files or not. Implementation safeguards can be put around operations, of course. The POC PR in zarr-python does not anticipate changes to the chunking structure at all... (which is enough for kerchunk to view archival data).

[d-v-b]

perhaps defining some semantics for "dirty" or "incomplete" zarr arrays would be useful. this would be a set of definitions (superset of zarr definitions) that define how to store temporary chunks. a library that recognizes these semantics would be capable of implementing an in-place rechunk by copying source chunks to a special location in the chunk hierarchy, deleting the source chunks, saving new chunks (from the data stored in the special location), then updating the metadata. if the state of the container at every point in this sequence was described formally via the "dirty" semantics, then this could be done safely.

[martindurant]

Certainly, I could imagine such a thing. I don't know if it needs to have a spec or just an implementation, but if the former, it would be a different ZEP (because chunking isn't the only way to make inconsistent structures).

[d-v-b]

agreed, and if anyone does produce such an implementation I would be super interested

[rabernat]

In terms of consistent updates, our current approach at Earthmover is to address this at the store level, via a transaction mechanism, rather than at the Zarr spec level. We have some ideas about how to migrate this to the spec.

However, I would encourage us to keep the scope of this ZEP a bit more narrow. Here we should just be figuring out how to present a view of variable chunk sizes to an application. How the data are generated can remain an implementation detail for now.

[tinaok]

When implementing domain decomposition for MPI applications, it's common to utilize 'optimized' variable chunk sizes. For instance, in the Earth System Model, when applying domain decomposition to the x-dimension (let's assume we have 100), with 7 MPI workers concurrently handling tasks, the standard MPI domain decomposition might be structured as 15, 15, 14, 14, 14, 14, 14. This allocation ensures that the worker with the most workload (15) and the one with the least (14) differ by only 1 task. This differs from a zarr chunk distribution, like 15, 15, 15, 15, 15, 15, 10, where the hardest worker (15) and the least burdened worker (10) differ by 5 tasks.

This optimized variable chunk sizing aims to minimize differences in execution times among MPI tasks as much as possible, considering that each MPI task necessitates periodic data exchanges.

I'm unable to employ kerchunk for these types of MPI Earth System Model outputs since kerchunk stop using xarray(dask) chunks as a backend and use zarr chunks.

[rabernat]

I was thinking through this ZEP the other day, and I had a thought about a possible change that I think could make things a lot simpler.

Right now, the chunks are encoded explicitly in the Array metadata

{
  "type": "rectangular", 
  "chunk_shape": [[5, 5, 5, 15, 15, 20, 35], 10],
  "separator":"/"
}

The chunks here will still be named the same way in the chunk store, i.e. c/0/0, c/1/0, c/2/0 etc. The only difference is that the chunks are not the same size.

But what if we did this instead.

{
  "type": "rectangular", 
  "chunk_shape": ["*", 10],
  "separator":"/"
}

where the * indicates that the chunk size is variable. Then we encode the chunk shape in the chunk name:

c/0-5/0
c/5-10/0
c/10-15/0
c/15-30/0
c/30-45/0
c/45-65/0
c/65-100/0

So we would essentially be moving information about the chunk shapes from the metadata doc to the store. To discover the chunk shape, an implementation would have to list the relevant directory of the chunk store.

---

Why do this? Here are some reasons:

• It keeps the metadata small.
• It resolves concurrency issues: multiple writers can now write different chunks concurrently in parallel more easily, because they don't have to synchronize their updates to the metadata doc
• It allows for some sparsity. We could have, for example

  c/0-5/0
  c/7-10/0
  

  where the chunk at 5-7 is implicitly missing
• Implementations could easily modify the chunking without touching the metadata, e.g. compact chunks.

Downsides:

• The listing operation may be more expensive than just reading the information from a json file. This is mitigated by using the hierarchical chunk naming default of V3, so that for N variable-sided chunks along a certain axis only requires listing N items in a single directory.
• Doesn't work for non-listable stores (but this is a problem for Zarr V3 more generally, not just this ZEP)

Thoughts?

[normanrz]

It's possible to list 50_000 items / s from S3 (ref). If there are 50_000 items in this JSON file, presumably it would also take some comparable time to fetch and parse. So I'm not sure that performance is a dealbreaker here.

The listing would probably need to be cached (which the json is already in most cases) to not have to fetch that for every single chunk.

HTTP which is quite popular for hosting shared data doesn't support listing and would be out entirely.

This is an issue for all of Zarr V3. There is currently no way to explore the hierarchy without being able to list the store.

Is it true that hierarchy discovery on HTTP is difficult. Reading arrays is not, though. Using the filenames to encode chunk sizes would break that for variable-chunked arrays.

I would favor a solution that is portable in a way that supports non-listable stores. Your are right that such a catalog or chunk manifest would require a coordinator.

[martindurant]

I will think about it, but one point.
the current proposal also allows for sparse/missing data: they will have a chunksize in the metadata but no corresponding chunk data file.

[ivirshup]

I'll also want to think about this a little longer. But would like to raise another point:

It resolves concurrency issues: multiple writers can now write different chunks concurrently in parallel more easily, because they don't have to synchronize their updates to the metadata doc

I think one of the big use cases for variable length chunks is when the size of each chunk is determined by the data, i.e. dynamic. The proposed of encoding chunk sizes requires that the chunk writer knows the length of all preceding chunks. In the current spec, a chunk writer could just know which order it's writing in (e.g. which URL it's writing to) so long as a coordinating process is handling finalizing the metadata.

This sorta rhymes with hive partitioning, but not quite since AFAIK that still uses order-based chunk naming at the lowest level and doesn't say anything about number of elements in a chunk. Is there other prior art for this approach?

[rabernat]

The proposed of encoding chunk sizes requires that the chunk writer knows the length of all preceding chunks.

Yes that's correct. I guess I assumed that would always be ok...the Zarr APIs I'm familiar with involve indexing into the full array when writing, i.e.

a[10:20] = new_data

...in which case the offsets are always known by the writing process. I guess that assumption is not true for your use case.

Is there other prior art for this approach?

Yes, this is kind of how TileDB works. Each write generates a fragment, which is an arbitrary range within the tile. When reading back the data, the reader scans all the fragments to figure out where their desired data lives. TileDB also includes a timestamp with each fragment, which allows multiple fragments to overlap the same range.

[jbms]

I think it would help to be clear about what the intended use cases are. We could imagine a TileDB-like format where there is no specific chunking, and you can write arbitrary rectangular regions (or possibly even non-rectangular regions) and then properly handle (and correctly order based on timestamp) overlapping writes at read time.

However, I think the use cases for such a format are sufficiently different from the existing zep 3 proposal, and the format has sufficiently-different trade-offs in terms of capabilities and efficiency, that it makes more sense to consider it separately from this proposal.

To be more specific:

• The original proposal has all of the chunk sizes stored directly in the metadata, which means they are known immediately when the array is opened, and requires relatively minor changes to the indexing logic. Storing the chunk sizes in a separate 1-d array would also not be very difficult to support, though you would presumably still have to just read the chunk size array in its entirety upon opening the array; partial reads of the chunk size array would be essentially impossible unless it was changed to cumulative sums, i.e. chunk partition points rather than chunk sizes, and even then there would be some challenges in supporting partially reading the chunk partition array efficiently.
• If more than the first dimension has chunking of "*", you can't just do a single (hierarchical) list operation to determine the chunk partition points. Instead you would have to (optionally-hierarchically) list everything up to the separator after the last "*"-chunked dimension, and then determine which partition points have been seen. For example, if the first two dimensions are "*"-chunked and are partitioned into m and n chunks, respectively, then compared to reading m + n chunk sizes from the metadata or separate chunk size arrays, we have to list m * n chunks.
• It would be possible for two chunks to overlap the same region. In this case, without a timestamp we would not know how to order them.
• Ultimately, if each top-level chunk used the sharding format, or a variant of it designed for variable-size chunks, and encoded in its key (1) a timestamp (2) the bounds of the region that it covers, and (3) optionally the latest key that also intersects this region on which the write is conditioned, then we would have something that is essentially the TileDB format, plus atomic read-modify-write for free as a bonus. This would allow arbitrary chunks without regard to any fixed grid, and would properly handle overlaps, but would have the same disadvantage as TileDB as far as read efficiency in the case of a large number of prior writes. Potentially there would be a way to choose the key encoding so that the normal lexicographical ordering of keys serves as a spatial index, such that when reading a given subregion, you could efficiently list the set of keys that overlaps it. That would mitigate the core limitation of TileDB. But I have not thought about this key encoding yet.

[alxmrs]

Another potential use case: Kerchunkifying/Virtualizing the Open-Meteo data format “OM”: fsspec/kerchunk#464