zarr-python

Zarr Python — 分块 N 维数组

概述

Zarr 是一个 Python 库，用于存储大型 N 维数组，支持分块、压缩和并行 I/O。它提供与 NumPy 兼容的索引方式，并支持可插拔的存储后端（本地、云端、内存），因此成为云原生科学数据管道的标准格式。

适用场景

• 通过分块访问存储大到无法放入内存的数组（外存计算）
• 使用 S3 或 GCS 存储后端构建云原生数据工作流
• 使用 Dask 并行读写，进行大规模计算
• 采用层次结构组织数据（包含带名称数组和元数据的组）
• 在不同格式之间转换（HDF5 → Zarr、NetCDF → Zarr）
• 以增量方式追加时间序列数据，无需重写
• 对于带标签且包含坐标的数组（时间、纬度、经度），请改用搭配 Zarr 后端的 xarray
• 对于数据管理、溯源和本体验证，请使用 lamindb（它使用 Zarr 作为存储格式）

先决条件

pip install zarr
# Cloud storage support
pip install s3fs   # Amazon S3
pip install gcsfs  # Google Cloud Storage

需要 Python 3.11 或更高版本。

快速入门

import zarr
import numpy as np

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

# Write with NumPy-style indexing
z[:, :] = np.random.random((10000, 10000)).astype("f4")

# Read a slice (only reads needed chunks)
subset = z[0:100, 0:100]
print(f"Shape: {subset.shape}, dtype: {subset.dtype}")
# Shape: (100, 100), dtype: float32

核心 API

1. 创建数组

import zarr
import numpy as np

# Empty arrays
z = zarr.zeros(shape=(10000, 10000), chunks=(1000, 1000), dtype="f4", store="data.zarr")
z = zarr.ones((5000, 5000), chunks=(500, 500), dtype="f4")
z = zarr.full((1000, 1000), fill_value=42, chunks=(100, 100), dtype="i4")

# From existing NumPy data
data = np.arange(10000, dtype="f4").reshape(100, 100)
z = zarr.array(data, chunks=(10, 10), store="from_numpy.zarr")
print(f"Created: shape={z.shape}, chunks={z.chunks}, dtype={z.dtype}")

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

# Open existing array
z = zarr.open_array("data.zarr", mode="r+")  # Read-write
z = zarr.open_array("data.zarr", mode="r")   # Read-only
z = zarr.open("data.zarr")                   # Auto-detect array vs group

2. 读取、写入和索引

import zarr
import numpy as np

z = zarr.zeros((10000, 10000), chunks=(1000, 1000), dtype="f4")

# Write slices
z[0, :] = np.arange(10000, dtype="f4")
z[10:20, 50:60] = np.random.random((10, 10)).astype("f4")
z[:] = 42  # Fill entire array

# Read slices (returns NumPy array)
row = z[5, :]
block = z[0:100, 0:100]
print(f"Row shape: {row.shape}, block shape: {block.shape}")

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

# Resize and append
z.resize(15000, 15000)
z.append(np.random.random((1000, 10000)).astype("f4"), axis=0)

3. 分块和分片

块形状是最重要的性能参数。

import zarr
from zarr.codecs import ShardingCodec

# Chunk aligned with access pattern
# Row-wise access → chunk spans columns
z_row = zarr.zeros((10000, 10000), chunks=(10, 10000), dtype="f4")

# Column-wise access → chunk spans rows
z_col = zarr.zeros((10000, 10000), chunks=(10000, 10), dtype="f4")

# Mixed access → balanced square chunks (~1MB each for float32)
z_bal = zarr.zeros((10000, 10000), chunks=(512, 512), dtype="f4")
# 512*512*4 bytes = ~1MB per chunk

# Sharding: group small chunks into larger storage objects
# Useful when millions of small chunks cause filesystem overhead
z_sharded = zarr.create_array(
    store="sharded.zarr",
    shape=(100000, 100000),
    chunks=(100, 100),       # Small chunks for fine-grained access
    shards=(1000, 1000),     # Groups 100 chunks per shard
    dtype="f4"
)
print(f"Chunks: {z_sharded.chunks}, shards reduce file count")

块大小指南：

• 每个块的目标大小为 1–10 MB（最小 1 MB）
• 让块形状与最常见的访问模式保持一致
• 读取时会将整个块载入内存 → 不要超过可用 RAM
• 写入时会将整个分片载入内存

4. 压缩

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

# Default: Blosc with Zstandard (good balance)
z = zarr.zeros((1000, 1000), chunks=(100, 100), dtype="f4")

# Explicit Blosc configuration
z = zarr.create_array(
    store="compressed.zarr",
    shape=(1000, 1000), chunks=(100, 100), dtype="f4",
    codecs=[BloscCodec(cname="zstd", clevel=5, shuffle="shuffle")]
)

# Speed-optimized (LZ4)
z_fast = zarr.create_array(
    store="fast.zarr",
    shape=(1000, 1000), chunks=(100, 100), dtype="f4",
    codecs=[BloscCodec(cname="lz4", clevel=1)]
)

# Maximum compression (Gzip level 9)
z_small = zarr.create_array(
    store="small.zarr",
    shape=(1000, 1000), chunks=(100, 100), dtype="f4",
    codecs=[GzipCodec(level=9)]
)

# No compression
z_raw = zarr.create_array(
    store="raw.zarr",
    shape=(1000, 1000), chunks=(100, 100), dtype="f4",
    codecs=[BytesCodec()]
)

编解码器选择：Blosc/Zstd（默认，均衡）→ LZ4（最快）→ Gzip（最小）。对于数值数据，请启用 shuffle="shuffle"——它会重新排列字节，以获得更高的压缩率。

5. 存储后端

import zarr
import numpy as np
from zarr.storage import LocalStore, MemoryStore, ZipStore

# Local filesystem (default — string paths create LocalStore automatically)
z = zarr.open_array("data/array.zarr", mode="w", shape=(1000, 1000),
                    chunks=(100, 100), dtype="f4")

# In-memory (not persisted)
store = MemoryStore()
z_mem = zarr.open_array(store=store, mode="w", shape=(1000, 1000),
                        chunks=(100, 100), dtype="f4")

# ZIP file storage
store = ZipStore("data.zip", mode="w")
z_zip = zarr.open_array(store=store, mode="w", shape=(1000, 1000),
                        chunks=(100, 100), dtype="f4")
z_zip[:] = np.random.random((1000, 1000)).astype("f4")
store.close()  # IMPORTANT: must close ZipStore

# Cloud storage: Amazon S3
import s3fs

s3 = s3fs.S3FileSystem(anon=False)
store = s3fs.S3Map(root="my-bucket/path/array.zarr", s3=s3)
z = zarr.open_array(store=store, mode="w", shape=(1000, 1000),
                    chunks=(500, 500), dtype="f4")
z[:] = np.random.random((1000, 1000)).astype("f4")

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

# Google Cloud Storage
import gcsfs
gcs = gcsfs.GCSFileSystem(project="my-project")
store = gcsfs.GCSMap(root="my-bucket/path/array.zarr", gcs=gcs)

云端最佳实践：合并元数据，使用 5–100 MB 的块，启用分片以减少对象数量，并使用 Dask 进行并行 I/O。

6. 组和层次结构

import zarr

# Create hierarchical structure (like HDF5 groups)
root = zarr.group(store="hierarchy.zarr")

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

# Create arrays within groups
temp_arr = temperature.create_array(
    name="t2m",
    shape=(365, 720, 1440),
    chunks=(1, 720, 1440),
    dtype="f4"
)
precip_arr = precipitation.create_array(
    name="prcp",
    shape=(365, 720, 1440),
    chunks=(1, 720, 1440),
    dtype="f4"
)

# Access by path
arr = root["temperature/t2m"]
print(root.tree())
# /
#  ├── temperature
#  │   └── t2m (365, 720, 1440) f4
#  └── precipitation
#      └── prcp (365, 720, 1440) f4

7. 属性和元数据

import zarr

z = zarr.zeros((1000, 1000), chunks=(100, 100), dtype="f4")

# Attach metadata (must be JSON-serializable)
z.attrs["description"] = "Temperature data in Kelvin"
z.attrs["units"] = "K"
z.attrs["processing_version"] = 2.1

print(z.attrs["units"])  # K

# Group-level attributes
root = zarr.group("data.zarr")
root.attrs["project"] = "Climate Analysis"
root.attrs["institution"] = "Research Institute"

核心概念

块大小选择指南

数据形状	访问模式	建议的块	原因
(N, M) 2D	按行访问	(小, M)	每个块覆盖整行
(N, M) 2D	按列访问	(N, 小)	每个块覆盖整列
(N, M) 2D	随机/混合	(√(1MB/dtype), √(1MB/dtype))	每个块约 1MB，保持均衡
(T, H, W) 时间序列	时间切片	(1, H, W)	每个块包含一个时间步
(T, H, W) 时间序列	空间区域	(T, 小, 小)	区域覆盖完整时间范围

1 MB 规则：对于 float32（4 字节），1 MB = 262,144 个元素。对于 float64（8 字节），1 MB = 131,072 个元素。

合并元数据

对于包含许多数组（10 个以上）的存储，将元数据合并为一次读取：

zarr.consolidate_metadata("data.zarr")
root = zarr.open_consolidated("data.zarr")  # Single metadata read

这对云存储至关重要（将 N 次元数据请求减少到 1 次）。注意：如果数组更新后未重新合并，合并的元数据就会过时。

常见工作流

工作流：云原生数据管道

import zarr
import numpy as np
import s3fs

# Step 1: Write to S3 with cloud-optimized chunks
s3 = s3fs.S3FileSystem()
store = s3fs.S3Map(root="s3://my-bucket/experiment.zarr", s3=s3)

root = zarr.group(store=store)
data_arr = root.create_array(
    name="measurements",
    shape=(10000, 10000),
    chunks=(500, 500),  # ~1MB chunks, good for cloud
    dtype="f4"
)
data_arr[:] = np.random.random((10000, 10000)).astype("f4")
data_arr.attrs["experiment"] = "batch_42"

# Step 2: Consolidate metadata
zarr.consolidate_metadata(store)

# Step 3: Read from anywhere
store_read = s3fs.S3Map(root="s3://my-bucket/experiment.zarr", s3=s3)
root_read = zarr.open_consolidated(store_read)
subset = root_read["measurements"][0:100, 0:100]
print(f"Read subset: {subset.shape}")

工作流：Dask 并行计算

import dask.array as da
import zarr
import numpy as np

# Step 1: Create large Zarr array
z = zarr.open("large_data.zarr", mode="w", shape=(100000, 100000),
              chunks=(1000, 1000), dtype="f4")
# (populate with data...)

# Step 2: Load as Dask array (lazy — no data loaded yet)
dask_arr = da.from_zarr("large_data.zarr")
print(f"Dask array: {dask_arr.shape}, {dask_arr.npartitions} partitions")

# Step 3: Compute in parallel (out-of-core)
col_means = dask_arr.mean(axis=0).compute()
print(f"Column means: {col_means.shape}")

# Step 4: Write Dask result back to Zarr
large_random = da.random.random((100000, 100000), chunks=(1000, 1000))
da.to_zarr(large_random, "output.zarr")

工作流：用于带标签数据的 Xarray-Zarr

import xarray as xr
import numpy as np
import pandas as pd

# Step 1: Create labeled dataset
ds = xr.Dataset(
    {
        "temperature": (["time", "lat", "lon"], np.random.random((365, 180, 360)).astype("f4")),
        "precipitation": (["time", "lat", "lon"], np.random.random((365, 180, 360)).astype("f4")),
    },
    coords={
        "time": pd.date_range("2024-01-01", periods=365),
        "lat": np.arange(-90, 90, 1.0),
        "lon": np.arange(-180, 180, 1.0),
    }
)

# Step 2: Save to Zarr
ds.to_zarr("climate.zarr")

# Step 3: Open with lazy loading
ds_loaded = xr.open_zarr("climate.zarr")
print(ds_loaded)

# Step 4: Label-based selection (only reads needed chunks)
subset = ds_loaded.sel(time="2024-06", lat=slice(30, 60))
print(f"June subset: {subset['temperature'].shape}")

工作流：格式转换

import zarr
import numpy as np

# HDF5 → Zarr
import h5py
with h5py.File("data.h5", "r") as h5:
    z = zarr.array(h5["dataset_name"][:], chunks=(1000, 1000), store="from_hdf5.zarr")

# NumPy → Zarr
data = np.load("data.npy")
z = zarr.array(data, chunks="auto", store="from_numpy.zarr")

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

关键参数

参数	模块	默认值	选项	作用
shape	create_array	必填	整数元组	数组维度
chunks	create_array	自动	整数元组、"auto"	各维度的分块形状
shards	create_array	无	整数元组	分片形状（将多个块分组）
dtype	create_array	"f8"	NumPy dtype	数据类型
codecs	create_array	Blosc/Zstd	编解码器对象列表	压缩流程
mode	open_array	"r"	"r"、"r+"、"w"、"a"	文件访问模式
store	全部	LocalStore	存储对象或路径	存储后端
cname	BloscCodec	"zstd"	"lz4"、"zstd"、"gzip" 等	压缩算法
clevel	BloscCodec	5	0–9	压缩级别
shuffle	BloscCodec	"noshuffle"	"shuffle"、"bitshuffle"	压缩前的字节重排

最佳实践

1. 每个块的大小设为 1–10 MB：块越小，元数据开销越大；块越大，每次读取所需的内存越多。对于 float32，512×512 ≈ 1 MB。

2. 让块与访问模式对齐：最常查询的维度应在块内连续存储。块未对齐时，性能差异可能高达 65 倍。

3. 反模式 — 使用 z[:] 加载整个数组：对于大型数组，请使用 Dask（da.from_zarr）或按明确指定的块进行处理。z[:] 会将所有数据加载到内存中。

4. 为云存储整合元数据：创建所有数组后，调用 zarr.consolidate_metadata(store)。然后使用 zarr.open_consolidated() 打开。这会将 N 次元数据读取减少到 1 次，对于 S3/GCS 的延迟至关重要。

5. 显式关闭 ZipStore：与其他存储不同，写入后必须对 ZipStore 调用 store.close()。忘记关闭会导致 ZIP 文件损坏。

6. 数百万个小块时使用分片：当块数超过约 100k 时，文件系统开销会成为主要瓶颈。分片会将多个块分组到较少的存储对象中。

7. 反模式 — 仅根据数组形状选择块形状：始终先考虑访问模式，再优化块大小。如果总是按行读取，那么看起来“合适”的形状（1000, 1000）可能很糟糕。

常见配方

配方：分析数组性能

import zarr

z = zarr.open("data.zarr")
print(z.info)
# Shows: type, shape, chunks, dtype, compressor, storage size

print(f"Compressed: {z.nbytes_stored / 1e6:.2f} MB")
print(f"Uncompressed: {z.nbytes / 1e6:.2f} MB")
print(f"Ratio: {z.nbytes / z.nbytes_stored:.1f}x")

配方：追加时间序列

import zarr
import numpy as np

# Create extensible array (start with 0 timesteps)
z = zarr.open("timeseries.zarr", mode="a",
              shape=(0, 720, 1440),
              chunks=(1, 720, 1440),
              dtype="f4")

# Append new timesteps incrementally
for day in range(365):
    new_step = np.random.random((1, 720, 1440)).astype("f4")
    z.append(new_step, axis=0)

print(f"Final shape: {z.shape}")  # (365, 720, 1440)

配方：使用 Dask 并行写入

import dask.array as da

# Generate large dataset in parallel
data = da.random.random((100000, 100000), chunks=(1000, 1000))

# Write to Zarr (parallel across chunks)
da.to_zarr(data, "parallel_output.zarr")

# Verify
z = da.from_zarr("parallel_output.zarr")
print(f"Written: {z.shape}, {z.npartitions} partitions")

故障排除

问题	原因	解决方案
读取性能缓慢	块形状与访问模式不对齐	分析访问模式；重新对齐块（按行访问 → 宽块）
读取时出现 MemoryError	加载整个数组或块过大	使用 Dask 的 da.from_zarr() 进行外存处理；减小块大小
云端延迟高	多次读取小型元数据	调用 zarr.consolidate_metadata(store)，然后调用 open_consolidated()
ZIP 存储损坏	忘记调用 store.close()	写入后务必关闭 ZipStore；使用上下文管理器
并发写入冲突	多个进程写入重叠的块	使用 ProcessSynchronizer，或确保写入的块互不重叠
压缩率低	数值数据未启用 shuffle	在 BloscCodec 中添加 shuffle="shuffle"
整合后的元数据过期	整合后修改了数组	更新后重新运行 zarr.consolidate_metadata()
ModuleNotFoundError: s3fs	缺少云存储依赖项	pip install s3fs（S3）或 pip install gcsfs（GCS）

相关技能

• lamindb-data-management — 使用 Zarr 作为存储后端；在 Zarr 数组之上提供数据管理、数据沿袭和本体验证
• anndata-data-structure — AnnData 对象可由 Zarr 存储支持，以便对单细胞数据进行外存处理
• sympy-symbolic-math — 无关，但同属科学计算类别

参考资料

• Zarr Python 文档 — 官方用户指南和 API 参考
• Zarr 规范 — 文件格式规范（V2、V3）
• Zarr GitHub — 源代码、问题跟踪和更新日志
• NumCodecs — Zarr 使用的压缩编解码器库