Skip to content

Zarr-Python 3.3.0

We're happy to announce the release of version 3.3.0 of Zarr-Python. It's been a while since our last release (3.2.1 dropped in May of this year), and we're bringing some exciting additions to the latest version. For the full release notes, see the 3.3.0 release notes, otherwise stick around for an overview of two performance-centric highlights of this release.

Faster low-latency storage

Relevant issues and pull requests:

  • #3524 -- the performance report that started this work
  • #3885 -- synchronous codec APIs and the FusedCodecPipeline

The cost of async overhead

Zarr-Python 3.x uses async routines for fetching data and decoding chunks. In terms of code, this means our store (data fetching) and codec (chunk decoding) APIs are both async. This makes I/O against high-latency storage backends like cloud object storage efficient. But for low-latency storage, like in-process memory or the file system, async routines add measurable overhead and offer no benefit. Async only adds value when there's work to be done while waiting for I/O to complete, but when I/O latency is low, it completes too quickly to run anything while waiting, and we are left paying the performance bill for obligatory async task scheduling that offered no value.

This performance problem became acute when Zarr-Python users reported that in-memory array indexing workloads ran slower in Zarr-Python 3.1.3 relative to Zarr-Python 2.18.7 (#3524). Fortunately this performance regression had a straightforward fix (I don't say "easy" because it was a lot of work).

Synchronous execution restores performance

If async overhead makes low-latency storage slow, does removing that overhead restore performance? Yes, it does!

In #3885 we defined synchronous versions of our storage and codec APIs -- the SyncByteGetter and SyncByteSetter protocols, plus a get_ranges_sync method on the Store ABC -- and then combined them in a new codec orchestration class called FusedCodecPipeline. The FusedCodecPipeline is an opt-in alternative to the default (the BatchedCodecPipeline) that gives large speedups for low-latency storage. It is currently marked experimental, so we may change it as we learn more; the default pipeline is untouched, and existing code keeps working unless you opt in.

The win here is not a faster compressor. It is the removal of async scheduling overhead (including some nasty asyncio.to_thread overhead), plus a few vectorized fast paths for dense, uncompressed shards. And we only expect this new pipeline to accelerate workloads targeting a subset of storage backends, namely any store with methods that advertise low latency.

On this author's 10-core Apple M4 laptop, the FusedCodecPipeline delivers the following results against memory-backed arrays:

  • uncompressed writes are ~4 times faster
  • uncompressed reads are ~5 times faster
  • compressed writes are ~2 times faster
  • compressed reads are ~2 times faster

These numbers came from a runnable example that ships with the documentation. Run it yourself to get a sense of how the FusedCodecPipeline behaves on your system -- when and how you use it depends on your hardware, your array layout, and how your chunks are compressed. What's certain is that for in-memory arrays, and arrays saved to the local file system, the FusedCodecPipeline is worth a try.

Getting good numbers requires choosing the right level of thread-based parallelism for your workload, which is part of the configuration of the FusedCodecPipeline. For uncompressed chunks there's no CPU-bound work to do after fetching a chunk and so thread-based parallelism is worse than useless and slows things down. But for compressed chunks, threading offers a substantial payoff.

How to use it

Select the pipeline through the runtime configuration by setting codec_pipeline.path. Set it globally to affect every array created or opened afterwards:

import zarr

zarr.config.set(
    {"codec_pipeline.path": "zarr.core.codec_pipeline.FusedCodecPipeline"}
)

Or scope it to a block of code by using zarr.config.set as a context manager, which is the safer choice if you only want the new pipeline for part of your program:

import numpy as np
import zarr
from zarr.storage import MemoryStore

with zarr.config.set(
    {"codec_pipeline.path": "zarr.core.codec_pipeline.FusedCodecPipeline"}
):
    arr = zarr.create_array(
        store=MemoryStore(),
        shape=(1000, 1000),
        chunks=(100, 100),
        shards=(1000, 1000),
        dtype="float32",
    )
    arr[:] = np.random.random((1000, 1000)).astype("float32")
    result = arr[:]

print(result.shape)

(1000, 1000)

Thread-based parallelism is configured separately, via codec_pipeline.max_workers. It defaults to None, meaning a pool sized to os.cpu_count(). Note that this setting is read only by the FusedCodecPipeline -- the default BatchedCodecPipeline ignores it, so tuning it without opting in above does nothing.

As noted, memory-backed and uncompressed workloads often do better with a single worker, which runs everything inline on the calling thread:

import zarr

# No thread pool: run codec compute inline. Often best for uncompressed,
# memory-backed arrays, where there's no CPU-bound work to overlap.
zarr.config.set({"codec_pipeline.max_workers": 1})

# A fixed-size thread pool, which pays off once compression is in play.
zarr.config.set({"codec_pipeline.max_workers": 8})

# Or back to the default, sized to the number of CPUs.
zarr.config.set({"codec_pipeline.max_workers": None})

To return to the default pipeline, set codec_pipeline.path back to the batched implementation:

import zarr

zarr.config.set(
    {"codec_pipeline.path": "zarr.core.codec_pipeline.BatchedCodecPipeline"}
)

Faster sharded reads

Relevant issues and pull requests:

  • #3004 -- optimize partial shard reads
  • #3925 -- Store.get_ranges for concurrent, coalesced multi-range reads
  • #3987 -- control coalescing through ArrayConfig and the runtime config

How sharding works

Chunks encoded with the sharding_indexed codec contain a secondary level of chunking, called subchunks. For example, if the chunk_grid field of the array metadata declares an "outer chunk" size of, say (10, 10), a sharding_indexed codec in the codecs field could declare an "inner chunk" size of (5, 5). Readers accessing such a chunk will observe a stored object (a stream of bytes) that decodes to an array with size (10, 10) (the "outer chunk"), which is comprised of four separate, contiguous byte ranges that each decode to a (5, 5) inner chunk. Each inner chunk occupies its own byte range in the outer chunk.

A reader can satisfy a request for all four inner chunks by issuing four separate byte-range requests, or by making a single request for a byte range that spans all four inner chunks. The latter option is nice because it cuts down on the number of requests we need. Historically Zarr-Python used this optimization when reading entire outer chunks; in 3.3.0, we use this optimization in more cases, resulting in more efficient I/O patterns for sharded reads.

Interval equivalence

Byte ranges, being intervals, obey some combination rules: the values in two half-open intervals [a, b), [b, c) can be captured by the single interval [a, c). That means a reader can get multiple inner chunks with one byte-range request by requesting a range of bytes starting with the first byte of the first subchunk and ending with the last byte of the last subchunk. When individual requests are expensive, this kind of optimization is worth a lot.

The requested inner chunks are not necessarily contiguous -- there might be a byte range gap between them. As long as that gap is not too big, its often efficient to fetch the entire byte range, gap included, and pick out the inner chunk byte ranges after I/O is done.

Byte range coalescing

We call this procedure -- merging adjacent byte ranges -- "byte range coalescing", and it's a new performance optimization shipping in Zarr-Python 3.3.0. Unlike the FusedCodecPipeline, this one is on by default with base settings we think are good, so most users won't need to tune anything.

Two knobs control it, both documented in the runtime configuration guide. Nearby byte ranges in the same shard are merged into a single request when the gap between them is no larger than array.sharding_coalesce_max_gap_bytes (default 1 MiB) and the merged read stays within array.sharding_coalesce_max_bytes (default 16 MiB). The gap threshold is what trades wasted bytes against saved requests: raising it reads more data you didn't ask for, in exchange for fewer requests.

For a runnable demonstration -- counting the store requests saved and timing them against a store with simulated latency -- see the sharded read coalescing example.

You can set them globally, or per array by passing config={...} to zarr.create_array:

import zarr
from zarr.storage import MemoryStore

arr = zarr.create_array(
    store=MemoryStore(),
    shape=(1000, 1000),
    chunks=(100, 100),
    shards=(1000, 1000),
    dtype="float32",
    config={
        "sharding_coalesce_max_gap_bytes": 4 * 1024**2,  # 4 MiB
        "sharding_coalesce_max_bytes": 64 * 1024**2,  # 64 MiB
    },
)
print(arr.shape)

(1000, 1000)

Tell us what you think

We hope these new features are helpful, and we would appreciate any feedback that helps us improve them, or any other aspect of Zarr-Python.

Going faster

The updates in this release are just the first step of a larger performance-oriented direction for Zarr-Python. Landing these two enhancements taught us a lot about the performance-sensitive areas of the library. We can and will invest more time in performance tuning, e.g. by adding or changing abstractions, writing code for special cases, etc.

We plan to consider including compiled code that should enable significant performance improvements. The zarrs project is an ecosystem of Zarr tools written in Rust, with extremely high performance. Is there a zarrs binding in Zarr-Python's future? I hope so! We are keenly observing development of zarrista as a proof-of-concept for what a Python-zarrs binding layer might look like. Stay tuned!