zarr-python

Par mkurman · zorai

Tableaux N-D fragmentés pour le stockage cloud. Tableaux compressés, I/O parallèle, intégration S3/GCS, compatibles NumPy/Dask/Xarray, pour les pipelines de calcul scientifique à grande échelle.

npx skills add https://github.com/mkurman/zorai --skill zarr-python

Zarr Python

Overview

Zarr est une bibliothèque Python pour stocker de grands tableaux N-dimensionnels avec découpage en chunks et compression. Appliquez cette compétence pour les E/S parallèles efficaces, les workflows cloud-native et l'intégration transparente avec NumPy, Dask et Xarray.

Quick Start

Installation

uv pip install zarr

Nécessite Python 3.11+. Pour le support du stockage cloud, installez des paquets supplémentaires :

uv pip install s3fs  # Pour S3
uv pip install gcsfs  # Pour Google Cloud Storage

Basic Array Creation

import zarr
import numpy as np

# Create a 2D array with chunking and compression
z = zarr.create_array(
    store="data/my_array.zarr",
    shape=(10000, 10000),
    chunks=(1000, 1000),
    dtype="f4"
)

# Write data using NumPy-style indexing
z[:, :] = np.random.random((10000, 10000))

# Read data
data = z[0:100, 0:100]  # Returns NumPy array

Core Operations

Creating Arrays

Zarr fournit plusieurs fonctions de commodité pour la création de tableaux :

# Create empty array
z = zarr.zeros(shape=(10000, 10000), chunks=(1000, 1000), dtype='f4',
               store='data.zarr')

# Create filled arrays
z = zarr.ones((5000, 5000), chunks=(500, 500))
z = zarr.full((1000, 1000), fill_value=42, chunks=(100, 100))

# Create from existing data
data = np.arange(10000).reshape(100, 100)
z = zarr.array(data, chunks=(10, 10), store='data.zarr')

# Create like another array
z2 = zarr.zeros_like(z)  # Matches shape, chunks, dtype of z

Opening Existing Arrays

# Open array (read/write mode by default)
z = zarr.open_array('data.zarr', mode='r+')

# Read-only mode
z = zarr.open_array('data.zarr', mode='r')

# The open() function auto-detects arrays vs groups
z = zarr.open('data.zarr')  # Returns Array or Group

Reading and Writing Data

Les tableaux Zarr supportent l'indexage de style NumPy :

# Write entire array
z[:] = 42

# Write slices
z[0, :] = np.arange(100)
z[10:20, 50:60] = np.random.random((10, 10))

# Read data (returns NumPy array)
data = z[0:100, 0:100]
row = z[5, :]

# Advanced indexing
z.vindex[[0, 5, 10], [2, 8, 15]]  # Coordinate indexing
z.oindex[0:10, [5, 10, 15]]       # Orthogonal indexing
z.blocks[0, 0]                     # Block/chunk indexing

Resizing and Appending

# Resize array
z.resize(15000, 15000)  # Expands or shrinks dimensions

# Append data along an axis
z.append(np.random.random((1000, 10000)), axis=0)  # Adds rows

Chunking Strategies

Le chunking est critique pour la performance. Choisissez les tailles et formes de chunks en fonction des patterns d'accès.

Chunk Size Guidelines

  • Taille minimum de chunk : 1 MB recommandé pour une performance optimale
  • Équilibre : Chunks plus grands = moins d'opérations de métadonnées; chunks plus petits = meilleur accès parallèle
  • Considération mémoire : Les chunks entiers doivent tenir en mémoire pendant la compression
# Configure chunk size (aim for ~1MB per chunk)
# For float32 data: 1MB = 262,144 elements = 512×512 array
z = zarr.zeros(
    shape=(10000, 10000),
    chunks=(512, 512),  # ~1MB chunks
    dtype='f4'
)

Aligning Chunks with Access Patterns

Critique : La forme du chunk affecte dramatiquement la performance selon la façon dont les données sont accédées.

# If accessing rows frequently (first dimension)
z = zarr.zeros((10000, 10000), chunks=(10, 10000))  # Chunk spans columns

# If accessing columns frequently (second dimension)
z = zarr.zeros((10000, 10000), chunks=(10000, 10))  # Chunk spans rows

# For mixed access patterns (balanced approach)
z = zarr.zeros((10000, 10000), chunks=(1000, 1000))  # Square chunks

Exemple de performance : Pour un tableau (200, 200, 200), lecture selon la première dimension :

  • Avec chunks (1, 200, 200) : ~107ms
  • Avec chunks (200, 200, 1) : ~1,65ms (65× plus rapide!)

Sharding for Large-Scale Storage

Quand les tableaux ont des millions de petits chunks, utilisez le sharding pour grouper les chunks dans de plus grands objets de stockage :

from zarr.codecs import ShardingCodec, BytesCodec
from zarr.codecs.blosc import BloscCodec

# Create array with sharding
z = zarr.create_array(
    store='data.zarr',
    shape=(100000, 100000),
    chunks=(100, 100),  # Small chunks for access
    shards=(1000, 1000),  # Groups 100 chunks per shard
    dtype='f4'
)

Avantages :

  • Réduit les frais système de fichiers liés à des millions de petits fichiers
  • Améliore la performance du stockage cloud (moins de requêtes d'objets)
  • Prévient le gaspillage de taille de bloc du système de fichiers

Important : Les shards entiers doivent tenir en mémoire avant l'écriture.

Compression

Zarr applique la compression par chunk pour réduire le stockage tout en maintenant un accès rapide.

Configuring Compression

from zarr.codecs.blosc import BloscCodec
from zarr.codecs import GzipCodec, ZstdCodec

# Default: Blosc with Zstandard
z = zarr.zeros((1000, 1000), chunks=(100, 100))  # Uses default compression

# Configure Blosc codec
z = zarr.create_array(
    store='data.zarr',
    shape=(1000, 1000),
    chunks=(100, 100),
    dtype='f4',
    codecs=[BloscCodec(cname='zstd', clevel=5, shuffle='shuffle')]
)

# Available Blosc compressors: 'blosclz', 'lz4', 'lz4hc', 'snappy', 'zlib', 'zstd'

# Use Gzip compression
z = zarr.create_array(
    store='data.zarr',
    shape=(1000, 1000),
    chunks=(100, 100),
    dtype='f4',
    codecs=[GzipCodec(level=6)]
)

# Disable compression
z = zarr.create_array(
    store='data.zarr',
    shape=(1000, 1000),
    chunks=(100, 100),
    dtype='f4',
    codecs=[BytesCodec()]  # No compression

Compression Performance Tips

  • Blosc (défaut) : Compression/décompression rapide, bon pour les workloads interactifs
  • Zstandard : Meilleurs ratios de compression, légèrement plus lent que LZ4
  • Gzip : Compression maximale, performance plus lente
  • LZ4 : Compression la plus rapide, ratios plus bas
  • Shuffle : Activez le filtre shuffle pour une meilleure compression sur les données numériques
# Optimal for numeric scientific data
codecs=[BloscCodec(cname='zstd', clevel=5, shuffle='shuffle')]

# Optimal for speed
codecs=[BloscCodec(cname='lz4', clevel=1)]

# Optimal for compression ratio
codecs=[GzipCodec(level=9)]

Storage Backends

Zarr supporte plusieurs backends de stockage via une interface de stockage flexible.

Local Filesystem (Default)

from zarr.storage import LocalStore

# Explicit store creation
store = LocalStore('data/my_array.zarr')
z = zarr.open_array(store=store, mode='w', shape=(1000, 1000), chunks=(100, 100))

# Or use string path (creates LocalStore automatically)
z = zarr.open_array('data/my_array.zarr', mode='w', shape=(1000, 1000),
                    chunks=(100, 100))

In-Memory Storage

from zarr.storage import MemoryStore

# Create in-memory store
store = MemoryStore()
z = zarr.open_array(store=store, mode='w', shape=(1000, 1000), chunks=(100, 100))

# Data exists only in memory, not persisted

ZIP File Storage

from zarr.storage import ZipStore

# Write to ZIP file
store = ZipStore('data.zip', mode='w')
z = zarr.open_array(store=store, mode='w', shape=(1000, 1000), chunks=(100, 100))
z[:] = np.random.random((1000, 1000))
store.close()  # IMPORTANT: Must close ZipStore

# Read from ZIP file
store = ZipStore('data.zip', mode='r')
z = zarr.open_array(store=store)
data = z[:]
store.close()

Cloud Storage (S3, GCS)

import s3fs
import zarr

# S3 storage
s3 = s3fs.S3FileSystem(anon=False)  # Use credentials
store = s3fs.S3Map(root='my-bucket/path/to/array.zarr', s3=s3)
z = zarr.open_array(store=store, mode='w', shape=(1000, 1000), chunks=(100, 100))
z[:] = data

# Google Cloud Storage
import gcsfs
gcs = gcsfs.GCSFileSystem(project='my-project')
store = gcsfs.GCSMap(root='my-bucket/path/to/array.zarr', gcs=gcs)
z = zarr.open_array(store=store, mode='w', shape=(1000, 1000), chunks=(100, 100))

Cloud Storage Best Practices :

  • Utilisez les métadonnées consolidées pour réduire la latence : zarr.consolidate_metadata(store)
  • Alignez les tailles de chunks avec le dimensionnement d'objets cloud (typiquement 5-100 MB optimal)
  • Activez les écritures parallèles en utilisant Dask pour les données à grande échelle
  • Envisagez le sharding pour réduire le nombre d'objets

Groups and Hierarchies

Les groupes organisent plusieurs tableaux hiérarchiquement, de manière similaire aux répertoires ou aux groupes HDF5.

Creating and Using Groups

# Create root group
root = zarr.group(store='data/hierarchy.zarr')

# Create sub-groups
temperature = root.create_group('temperature')
precipitation = root.create_group('precipitation')

# Create arrays within groups
temp_array = temperature.create_array(
    name='t2m',
    shape=(365, 720, 1440),
    chunks=(1, 720, 1440),
    dtype='f4'
)

precip_array = precipitation.create_array(
    name='prcp',
    shape=(365, 720, 1440),
    chunks=(1, 720, 1440),
    dtype='f4'
)

# Access using paths
array = root['temperature/t2m']

# Visualize hierarchy
print(root.tree())
# Output:
# /
#  ├── temperature
#  │   └── t2m (365, 720, 1440) f4
#  └── precipitation
#      └── prcp (365, 720, 1440) f4

H5py-Compatible API

Zarr fournit une interface compatible h5py pour les utilisateurs familiers de HDF5 :

# Create group with h5py-style methods
root = zarr.group('data.zarr')
dataset = root.create_dataset('my_data', shape=(1000, 1000), chunks=(100, 100),
                              dtype='f4')

# Access like h5py
grp = root.require_group('subgroup')
arr = grp.require_dataset('array', shape=(500, 500), chunks=(50, 50), dtype='i4')

Attributes and Metadata

Attachez des métadonnées personnalisées aux tableaux et groupes à l'aide d'attributs :

# Add attributes to array
z = zarr.zeros((1000, 1000), chunks=(100, 100))
z.attrs['description'] = 'Temperature data in Kelvin'
z.attrs['units'] = 'K'
z.attrs['created'] = '2024-01-15'
z.attrs['processing_version'] = 2.1

# Attributes are stored as JSON
print(z.attrs['units'])  # Output: K

# Add attributes to groups
root = zarr.group('data.zarr')
root.attrs['project'] = 'Climate Analysis'
root.attrs['institution'] = 'Research Institute'

# Attributes persist with the array/group
z2 = zarr.open('data.zarr')
print(z2.attrs['description'])

Important : Les attributs doivent être sérialisables en JSON (chaînes, nombres, listes, dicts, booléens, null).

Integration with NumPy, Dask, and Xarray

NumPy Integration

Les tableaux Zarr implémentent l'interface de tableau NumPy :

import numpy as np
import zarr

z = zarr.zeros((1000, 1000), chunks=(100, 100))

# Use NumPy functions directly
result = np.sum(z, axis=0)  # NumPy operates on Zarr array
mean = np.mean(z[:100, :100])

# Convert to NumPy array
numpy_array = z[:]  # Loads entire array into memory

Dask Integration

Dask fournit un calcul lazy et parallèle sur les tableaux Zarr :

import dask.array as da
import zarr

# Create large Zarr array
z = zarr.open('data.zarr', mode='w', shape=(100000, 100000),
              chunks=(1000, 1000), dtype='f4')

# Load as Dask array (lazy, no data loaded)
dask_array = da.from_zarr('data.zarr')

# Perform computations (parallel, out-of-core)
result = dask_array.mean(axis=0).compute()  # Parallel computation

# Write Dask array to Zarr
large_array = da.random.random((100000, 100000), chunks=(1000, 1000))
da.to_zarr(large_array, 'output.zarr')

Avantages :

  • Traiter les datasets plus volumineux que la mémoire
  • Calcul parallèle automatique sur les chunks
  • E/S efficace avec le stockage chunked

Xarray Integration

Xarray fournit des tableaux multidimensionnels étiquetés avec backend Zarr :

import xarray as xr
import zarr

# Open Zarr store as Xarray Dataset (lazy loading)
ds = xr.open_zarr('data.zarr')

# Dataset includes coordinates and metadata
print(ds)

# Access variables
temperature = ds['temperature']

# Perform labeled operations
subset = ds.sel(time='2024-01', lat=slice(30, 60))

# Write Xarray Dataset to Zarr
ds.to_zarr('output.zarr')

# Create from scratch with coordinates
ds = xr.Dataset(
    {
        'temperature': (['time', 'lat', 'lon'], data),
        'precipitation': (['time', 'lat', 'lon'], data2)
    },
    coords={
        'time': pd.date_range('2024-01-01', periods=365),
        'lat': np.arange(-90, 91, 1),
        'lon': np.arange(-180, 180, 1)
    }
)
ds.to_zarr('climate_data.zarr')

Avantages :

  • Dimensions et coordonnées nommées
  • Indexation et sélection basées sur les étiquettes
  • Intégration avec pandas pour les séries temporelles
  • Interface de type NetCDF familière aux climatologues et géospécialistes

Parallel Computing and Synchronization

Thread-Safe Operations

from zarr import ThreadSynchronizer
import zarr

# For multi-threaded writes
synchronizer = ThreadSynchronizer()
z = zarr.open_array('data.zarr', mode='r+', shape=(10000, 10000),
                    chunks=(1000, 1000), synchronizer=synchronizer)

# Safe for concurrent writes from multiple threads
# (when writes don't span chunk boundaries)

Process-Safe Operations

from zarr import ProcessSynchronizer
import zarr

# For multi-process writes
synchronizer = ProcessSynchronizer('sync_data.sync')
z = zarr.open_array('data.zarr', mode='r+', shape=(10000, 10000),
                    chunks=(1000, 1000), synchronizer=synchronizer)

# Safe for concurrent writes from multiple processes

Note :

  • Les lectures concurrentes ne nécessitent aucune synchronisation
  • La synchronisation est nécessaire uniquement pour les écritures qui peuvent traverser les limites de chunk
  • Chaque processus/thread écrivant dans des chunks séparés ne nécessite aucune synchronisation

Consolidated Metadata

Pour les stores hiérarchiques avec de nombreux tableaux, consolidez les métadonnées dans un fichier unique pour réduire les opérations E/S :

import zarr

# After creating arrays/groups
root = zarr.group('data.zarr')
# ... create multiple arrays/groups ...

# Consolidate metadata
zarr.consolidate_metadata('data.zarr')

# Open with consolidated metadata (faster, especially on cloud storage)
root = zarr.open_consolidated('data.zarr')

Avantages :

  • Réduit les opérations de lecture de métadonnées de N (une par tableau) à 1
  • Critique pour le stockage cloud (réduit la latence)
  • Accélère les opérations tree() et la traversée de groupes

Précautions :

  • Les métadonnées peuvent devenir obsolètes si les tableaux se mettent à jour sans re-consolidation
  • Non adapté aux datasets fréquemment mis à jour
  • Les scénarios multi-escriteurs peuvent avoir des lectures incohérentes

Performance Optimization

Checklist for Optimal Performance

  1. Chunk Size: Visez 1-10 MB par chunk

    # For float32: 1MB = 262,144 elements
chunks = (512, 512)  # 512×512×4 bytes = ~1MB

  2. Chunk Shape: Alignez avec les patterns d'accès

    # Row-wise access → chunk spans columns: (small, large)
# Column-wise access → chunk spans rows: (large, small)
# Random access → balanced: (medium, medium)

  3. Compression: Choisissez selon le workload

    # Interactive/fast: BloscCodec(cname='lz4')
# Balanced: BloscCodec(cname='zstd', clevel=5)
# Maximum compression: GzipCodec(level=9)

  4. Storage Backend: Adaptez à l'environnement

    # Local: LocalStore (default)
# Cloud: S3Map/GCSMap with consolidated metadata
# Temporary: MemoryStore

  5. Sharding: Utilisez pour les datasets à grande échelle

    # When you have millions of small chunks
shards=(10*chunk_size, 10*chunk_size)

  6. Parallel I/O: Utilisez Dask pour les grandes opérations

    import dask.array as da
dask_array = da.from_zarr('data.zarr')
result = dask_array.compute(scheduler='threads', num_workers=8)

Profiling and Debugging

# Print detailed array information
print(z.info)

# Output includes:
# - Type, shape, chunks, dtype
# - Compression codec and level
# - Storage size (compressed vs uncompressed)
# - Storage location

# Check storage size
print(f"Compressed size: {z.nbytes_stored / 1e6:.2f} MB")
print(f"Uncompressed size: {z.nbytes / 1e6:.2f} MB")
print(f"Compression ratio: {z.nbytes / z.nbytes_stored:.2f}x")

Common Patterns and Best Practices

Pattern: Time Series Data

# Store time series with time as first dimension
# This allows efficient appending of new time steps
z = zarr.open('timeseries.zarr', mode='a',
              shape=(0, 720, 1440),  # Start with 0 time steps
              chunks=(1, 720, 1440),  # One time step per chunk
              dtype='f4')

# Append new time steps
new_data = np.random.random((1, 720, 1440))
z.append(new_data, axis=0)

Pattern: Large Matrix Operations

import dask.array as da

# Create large matrix in Zarr
z = zarr.open('matrix.zarr', mode='w',
              shape=(100000, 100000),
              chunks=(1000, 1000),
              dtype='f8')

# Use Dask for parallel computation
dask_z = da.from_zarr('matrix.zarr')
result = (dask_z @ dask_z.T).compute()  # Parallel matrix multiply

Pattern: Cloud-Native Workflow

import s3fs
import zarr

# Write to S3
s3 = s3fs.S3FileSystem()
store = s3fs.S3Map(root='s3://my-bucket/data.zarr', s3=s3)

# Create array with appropriate chunking for cloud
z = zarr.open_array(store=store, mode='w',
                    shape=(10000, 10000),
                    chunks=(500, 500),  # ~1MB chunks
                    dtype='f4')
z[:] = data

# Consolidate metadata for faster reads
zarr.consolidate_metadata(store)

# Read from S3 (anywhere, anytime)
store_read = s3fs.S3Map(root='s3://my-bucket/data.zarr', s3=s3)
z_read = zarr.open_consolidated(store_read)
subset = z_read[0:100, 0:100]

Pattern: Format Conversion

# HDF5 to Zarr
import h5py
import zarr

with h5py.File('data.h5', 'r') as h5:
    dataset = h5['dataset_name']
    z = zarr.array(dataset[:],
                   chunks=(1000, 1000),
                   store='data.zarr')

# NumPy to Zarr
import numpy as np
data = np.load('data.npy')
z = zarr.array(data, chunks='auto', store='data.zarr')

# Zarr to NetCDF (via Xarray)
import xarray as xr
ds = xr.open_zarr('data.zarr')
ds.to_netcdf('data.nc')

Common Issues and Solutions

Issue: Slow Performance

Diagnosis: Vérifiez la taille et l'alignement du chunk

print(z.chunks)  # Are chunks appropriate size?
print(z.info)    # Check compression ratio

Solutions :

  • Augmentez la taille du chunk à 1-10 MB
  • Alignez les chunks avec le pattern d'accès
  • Essayez différents codecs de compression
  • Utilisez Dask pour les opérations parallèles

Issue: High Memory Usage

Cause : Chargement du tableau entier ou de grands chunks en mémoire

Solutions :

# Don't load entire array
# Bad: data = z[:]
# Good: Process in chunks
for i in range(0, z.shape[0], 1000):
    chunk = z[i:i+1000, :]
    process(chunk)

# Or use Dask for automatic chunking
import dask.array as da
dask_z = da.from_zarr('data.zarr')
result = dask_z.mean().compute()  # Processes in chunks

Issue: Cloud Storage Latency

Solutions :

# 1. Consolidate metadata
zarr.consolidate_metadata(store)
z = zarr.open_consolidated(store)

# 2. Use appropriate chunk sizes (5-100 MB for cloud)
chunks = (2000, 2000)  # Larger chunks for cloud

# 3. Enable sharding
shards = (10000, 10000)  # Groups many chunks

Issue: Concurrent Write Conflicts

Solution : Utilisez des synchroniseurs ou assurez-vous que les écritures ne se chevauchent pas

from zarr import ProcessSynchronizer

sync = ProcessSynchronizer('sync.sync')
z = zarr.open_array('data.zarr', mode='r+', synchronizer=sync)

# Or design workflow so each process writes to separate chunks

Additional Resources

Pour la documentation détaillée de l'API, l'utilisation avancée et les dernières mises à jour :

Related Libraries :

Skills similaires

dask
mkurman / zorai

Traiter des données massives en parallèle avec Dask sur CPU unique ou cluster distribué.

309
get-available-resources
mkurman / zorai

Détecter les ressources CPU, GPU et mémoire pour optimiser les tâches de calcul scientifique.

309
accelerated-computing-cudf
nvidia / skills

Accélérer des DataFrames pandas sur GPU avec cuDF et dask-cuDF.

1 033
earth2studio-data-fetch
nvidia / skills

Télécharger des données météo/climat via les APIs Earth2Studio avec vérification lexicale.

1 033
anndata
mkurman / zorai

Manipuler des matrices de données annotées pour l'analyse génomique single-cell.

309