zarr-python

Zarr Python — チャンク化された N 次元配列

概要

Zarr は、チャンク化、圧縮、並列 I/O を使用して、大規模な N 次元配列を保存するための Python ライブラリです。プラグイン式のストレージバックエンド（ローカル、クラウド、インメモリ）を備え、NumPy 互換のインデックス指定を提供するため、クラウドネイティブな科学データパイプラインの標準フォーマットとなっています。

使用する場面

• チャンク単位でアクセスしながら、メモリに収まらない大きさの配列を保存する場合（アウトオブコアコンピューティング）
• S3 または GCS のストレージバックエンドを使ったクラウドネイティブなデータワークフロー
• 大規模な計算で Dask を使用して並列読み取り／書き込みを行う場合
• 階層的なデータ構成（メタデータ付きの名前付き配列グループ）
• フォーマット間で変換する場合（HDF5 → Zarr、NetCDF → Zarr）
• 時系列データを、全体を書き直さずに段階的に追加する場合
• ラベル付きの座標対応配列（time、lat、lon）には、代わりに Zarr バックエンドを使用する xarray をお使いください
• データ管理、リネージ、オントロジー検証には、Zarr をストレージフォーマットとして使用する lamindb をお使いください

前提条件

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 のチャンクを使用し、オブジェクト数を減らすためにシャーディングを有効にし、並列 I/O には Dask を使用します。

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)	1 チャンクあたり 1 タイムステップです
(T, H, W) 時系列	空間領域	(T, 小, 小)	領域の全時間を含みます

1 MB のルール：float32（4 バイト）では、1 MB = 262,144 要素です。float64（8 バイト）では、1 MB = 131,072 要素です。

統合メタデータ

多数の配列（10 個以上）を含むストアでは、メタデータを 1 回の読み取りに統合します。

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 のデータ型	データ型
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 を使うか、重複しないチャンクに書き込むようにします
圧縮率が低い	数値データにシャッフルが設定されていない	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 — 関連はありませんが、scientific-computingカテゴリに共存しています

参考資料

• Zarr Python ドキュメント — 公式ユーザーガイドおよび API リファレンス
• Zarr 仕様 — ファイル形式の仕様（V2、V3）
• Zarr GitHub — ソースコード、課題、変更履歴
• NumCodecs — Zarr が使用する圧縮コーデックライブラリ