Skip to content

Experimental features

This section contains documentation for experimental Zarr Python features. The features described here are exciting and potentially useful, but also volatile -- we might change them at any time. Take this into account if you consider depending on these features. See the experimental API policy for the stability guarantees (or lack thereof) that apply to everything documented on this page.

FusedCodecPipeline

A codec pipeline is the machinery that turns chunks of array data into stored bytes and back, by running the configured codecs (filters, serializer, compressors) and performing the storage IO. The default pipeline, BatchedCodecPipeline, schedules both the IO and codec work asynchronously -- roughly one coroutine per chunk operation.

FusedCodecPipeline is an experimental alternative that runs codec compute and synchronous IO synchronously, avoiding that per-chunk async scheduling overhead and nasty asyncio.to_thread overhead. On real workloads the scheduling cost can dominate the actual codec work, so removing it is a significant speedup -- especially for sharded arrays, where a single shard read or write involves many inner chunks.

Note: The win is not a faster compressor or a different on-disk format -- the bytes written are identical. It is purely the removal of async scheduling overhead, plus a few vectorized fast paths for dense, uncompressed shards i.e., removing compute where it is not needed.

When it helps

There are two main benefits in this new pipeline:

  1. When storage IO is fast enough that the scheduling overhead, not the IO itself, is the bottleneck. That means low-latency stores that are themselves synchronous -- in particular zarr.storage.MemoryStore and zarr.storage.LocalStore.

  2. Whenever codec work that is truly synchronous will not need the overhead of async scheduling i.e., inner-chunk codec work in sharding using something like zstd. We also now make use of asyncio.as_completed so that IO from asynchronous sources can begin decompression immediately.

Opting in

FusedCodecPipeline is opt-in: the default pipeline is unchanged, so existing code behaves exactly as before. Select it through the runtime configuration, by setting codec_pipeline.path:

import zarr

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

You can set this globally as above (affecting every array created or opened afterwards), or scope it to a block of code using zarr.config.set as a context manager:

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

with zarr.config.set(
    {"codec_pipeline.path": "zarr.core.codec_pipeline.FusedCodecPipeline"}
):
    # A sharded array on an in-memory store -- the low-latency case the
    # synchronous pipeline targets.
    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)

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"}
)

Threading

By default the synchronous pipeline runs fully threaded i.e., os.cpu_count(). For memory-backed workflows, you may find that setting max_workers to 1 helps (since requests for data from the store are GIL-locked, unlike, say, file-backed i/o).

import zarr

# Use a fixed-size thread pool for codec compute.
zarr.config.set({"codec_pipeline.max_workers": 8})

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

On many-core nodes a pool sized to cpu_count can oversubscribe workloads that already parallelize at a higher level (e.g. Dask). codec_pipeline.max_workers only affects FusedCodecPipeline; the default BatchedCodecPipeline ignores it.

CacheStore

Zarr Python 3.1.4 adds zarr.experimental.cache_store.CacheStore, which provides a dual-store caching implementation that can be wrapped around any Zarr store to improve performance for repeated data access. This is particularly useful when working with remote stores (e.g., S3, HTTP) where network latency can significantly impact data access speed.

The CacheStore implements a cache that uses a separate Store instance as the cache backend, providing persistent caching capabilities with time-based expiration, size-based eviction, and flexible cache storage options. It automatically evicts the least recently used items when the cache reaches its maximum size.

Because the CacheStore uses an ordinary Zarr Store object as the caching layer, you can reuse the data stored in the cache later.

Note

The CacheStore is a wrapper store that maintains compatibility with the full zarr.abc.store.Store API while adding transparent caching functionality.

Basic Usage

Creating a CacheStore requires both a source store and a cache store. The cache store can be any Store implementation, providing flexibility in cache persistence:

import zarr
from zarr.storage import LocalStore
import numpy as np
from tempfile import mkdtemp
from zarr.experimental.cache_store import CacheStore

# Create a local store and a separate cache store
local_store_path = mkdtemp(suffix='.zarr')
source_store = LocalStore(local_store_path)
cache_store = zarr.storage.MemoryStore()  # In-memory cache
cached_store = CacheStore(
    store=source_store,
    cache_store=cache_store,
    max_size=256*1024*1024  # 256MB cache
)

# Create an array using the cached store
zarr_array = zarr.zeros((100, 100), chunks=(10, 10), dtype='f8', store=cached_store, mode='w')

# Write some data to force chunk creation
zarr_array[:] = np.random.random((100, 100))

The dual-store architecture allows you to use different store types for source and cache, such as a remote store for source data and a local store for persistent caching.

Performance Benefits

The CacheStore provides significant performance improvements for repeated data access:

import time

# Benchmark reading with cache
start = time.time()
for _ in range(100):
    _ = zarr_array[:]
elapsed_cache = time.time() - start

# Compare with direct store access (without cache)
zarr_array_nocache = zarr.open(local_store_path, mode='r')
start = time.time()
for _ in range(100):
    _ = zarr_array_nocache[:]
elapsed_nocache = time.time() - start

# Cache provides speedup for repeated access
speedup = elapsed_nocache / elapsed_cache
print(f"Speedup is {speedup}")

Speedup is 2.417225885209755

Cache effectiveness is particularly pronounced with repeated access to the same data chunks.

Cache Configuration

The CacheStore can be configured with several parameters:

max_size: Controls the maximum size of cached data in bytes. The Basic Usage example above sets a 256MB limit with max_size=256*1024*1024:

# Unlimited cache size (use with caution)
cache = CacheStore(
    store=source_store,
    cache_store=cache_store,
    max_size=None
)

max_age_seconds: Controls time-based cache expiration

# Cache expires after 1 hour
cache = CacheStore(
    store=source_store,
    cache_store=cache_store,
    max_age_seconds=3600
)

# Cache never expires
cache = CacheStore(
    store=source_store,
    cache_store=cache_store,
    max_age_seconds="infinity"
)

cache_set_data: Controls whether written data is cached

# Cache data when writing (default)
cache = CacheStore(
    store=source_store,
    cache_store=cache_store,
    cache_set_data=True
)

# Don't cache written data (read-only cache)
cache = CacheStore(
    store=source_store,
    cache_store=cache_store,
    cache_set_data=False
)

Cache Statistics

The CacheStore provides statistics to monitor cache performance and state:

# Access some data to generate cache activity
# (these chunks were already cached by the reads above, so both accesses are cache hits)
data = zarr_array[0:50, 0:50]
data = zarr_array[0:50, 0:50]

# Get comprehensive cache information
info = cached_store.cache_info()
print(info['cache_store_type'])  # e.g., 'MemoryStore'
print(info['max_age_seconds'])
print(info['max_size'])
print(info['current_size'])
print(info['tracked_keys'])
print(info['cached_keys'])
print(info['cache_set_data'])

MemoryStore infinity 268435456 81641 101 101 True

The cache_info() method returns a dictionary with detailed information about the cache state.

Cache Management

The CacheStore provides methods for manual cache management:

# Clear all cached data and tracking information
import asyncio
asyncio.run(cached_store.clear_cache())

# Check cache info after clearing
info = cached_store.cache_info()
assert info['tracked_keys'] == 0
assert info['current_size'] == 0

The clear_cache() method is an async method that clears both the cache store (if it supports the clear method) and all internal tracking data.

Best Practices

  1. Choose appropriate cache store: Use MemoryStore for fast temporary caching or LocalStore for persistent caching
  2. Size the cache appropriately: Set max_size based on available storage and expected data access patterns
  3. Use with remote stores: The cache provides the most benefit when wrapping slow remote stores
  4. Monitor cache statistics: Use cache_info() to tune cache size and access patterns
  5. Consider data locality: Group related data accesses together to improve cache efficiency
  6. Set appropriate expiration: Use max_age_seconds for time-sensitive data or "infinity" for static data

Working with Different Store Types

The CacheStore can wrap any store that implements the zarr.abc.store.Store interface and use any store type for the cache backend:

Local Store with Memory Cache

from zarr.storage import LocalStore, MemoryStore
from zarr.experimental.cache_store import CacheStore
from tempfile import mkdtemp

local_store_path = mkdtemp(suffix='.zarr')
source_store = LocalStore(local_store_path)
cache_store = MemoryStore()
cached_store = CacheStore(
    store=source_store,
    cache_store=cache_store,
    max_size=128*1024*1024
)

Memory Store with Persistent Cache

from tempfile import mkdtemp
from zarr.storage import MemoryStore, LocalStore
from zarr.experimental.cache_store import CacheStore

memory_store = MemoryStore()
local_store_path = mkdtemp(suffix='.zarr')
persistent_cache = LocalStore(local_store_path)
cached_store = CacheStore(
    store=memory_store,
    cache_store=persistent_cache,
    max_size=256*1024*1024
)

The dual-store architecture provides flexibility in choosing the best combination of source and cache stores for your specific use case.

Examples from Real Usage

Here's a complete example demonstrating cache effectiveness:

import numpy as np
import time
from tempfile import mkdtemp
import zarr
import zarr.storage
from zarr.experimental.cache_store import CacheStore

# Create test data with dual-store cache
local_store_path = mkdtemp(suffix='.zarr')
source_store = zarr.storage.LocalStore(local_store_path)
cache_store = zarr.storage.MemoryStore()
cached_store = CacheStore(
    store=source_store,
    cache_store=cache_store,
    max_size=256*1024*1024
)
zarr_array = zarr.zeros((100, 100), chunks=(10, 10), dtype='f8', store=cached_store, mode='w')
zarr_array[:] = np.random.random((100, 100))

# Demonstrate cache effectiveness with repeated access
start = time.time()
data = zarr_array[20:30, 20:30]  # First access (cache miss)
first_access = time.time() - start
print(f"First access took {first_access}")

start = time.time()
data = zarr_array[20:30, 20:30]  # Second access (cache hit)
second_access = time.time() - start
print(f"Second access took {second_access}")

# Check cache statistics
info = cached_store.cache_info()
assert info['cached_keys'] > 0  # Should have cached keys
assert info['current_size'] > 0  # Should have cached data
print(f"Cache contains {info['cached_keys']} keys with {info['current_size']} bytes")

First access took 0.00083160400390625 Second access took 0.0007557868957519531 Cache contains 101 keys with 81641 bytes

This example shows how the CacheStore can significantly reduce access times for repeated data reads, particularly important when working with remote data sources. The dual-store architecture allows for flexible cache persistence and management.