Set of tools for compressing netCDF files with Zarr.

The tools use the following compression libraries:

• Numcodecs: Zarr native library [documentation]
• EBCC (optional): Error Bounded Climate Compressor, see EBCC below

Installation

System Prerequisites

• C/C++ compiler toolchain (required to build mpi4py)
• MPI implementation (required for mpi4py)
• ecCodes library for GRIB files

On Santis@ALPS:

export UENV_NAME="prgenv-gnu/26.3:v1"

On Balfrin@ALPS:

export UENV_NAME="netcdf-tools/2024:v1"

Then:

uenv image pull $UENV_NAME
uenv start --view=default $UENV_NAME

once the above is complete (the uenv steps are for ALPS only; locally they are not needed):

git clone git@github.com:C2SM/data-compression.git dc_toolkit
cd dc_toolkit
rm -rf venv
python -m venv venv
source venv/bin/activate
bash install_dc_toolkit.sh

On Santis, DC_CONSTRAINTS=constraints/santis.txt bash install_dc_toolkit.sh installs the dependency versions the production runs were validated with.

Tests: pip install -e ".[test]", then pytest -m "not mpi and not slow"; the mpi tests start dc_toolkit through DC_TOOLKIT_MPIEXEC (e.g. "srun --ntasks={n} --cpus-per-task=1" inside an allocation, else mpiexec -n {n}).

Usage

New to the toolkit? docs/intro.md goes from a laptop installation to a verified store: both commands and their pipelines, compress without a sweep, and EBCC.

--------------------------------------------------------------------------------

Usage: dc_toolkit --help           # List of available commands
Usage: dc_toolkit COMMAND --help   # Documentation per command

Example:

dc_toolkit \                                                # CLI-tool
  evaluate_combos \                                         # command
  netCDF_files/tigge_pl_t_q_dx=2_2024_08_02.nc \            # netCDF file
  --where-to-write ./dump \                                 # output directory
  --field-to-compress t \                                   # field to sweep
  --l1-threshold 0.005 \                                    # relative L1 budget (0.5 %), required
  --eval-data-size-limit 5GB                                # sample size

--------------------------------------------------------------------------------

End-to-end workflow

The typical pipeline is two commands:

1. evaluate_combos — sweep (compressor × filter × serializer) combinations on a representative sample of each field and record the compression ratio and error metrics of every combo. --l1-threshold (a relative L1 error budget) is mandatory; the L2, Linf and bias gates default to 2x, 10x and 0.5x of it. The winner of each field goes to manifest_{var}.json. With --resume (the default) combos already recorded in the output directory are not evaluated again: their metrics are reused (as long as the dataset, the sample, the chunk settings, the physical bounds, the code that measures them and the row layout, the library versions and EBCC's env vars are unchanged), the gates are re-applied with the current thresholds, a combo that lacks a metric a newly enabled gate needs is evaluated again (and fails that gate if its re-evaluation fails), and rows outside the current codec space are left out of the results. A combo in flight when a rank died (a segfault, the OOM killer) is evaluated alone on the next run, after the others, and left out for good if it kills the rank again; a cancelled or timed-out job blames no combo. --no-resume, or a change to what the rows were measured on, starts the field from scratch and keeps its previous results, manifest and plan as *.previous. Two sweeps of one field into one directory are refused (sweep_{var}.lock). --phys-min / --phys-max reject a combo that moves a cell from within the bounds to beyond them; cells the source already has beyond a bound (sentinels, a field that dips below 0) do not count. --compressor-class / --filter-class / --serializer-class accept a fixed list of names (a typo is refused when the command line is parsed); a field whose dtype the chosen class cannot take is skipped with a message (an error when it is the --field-to-compress), except integer fields, which have Delta as their only filter and get it for any --filter-class but none (with a message when the class names another filter); variables that are not numeric arrays (datetimes, strings, scalars such as crs) are skipped too, and so are CF bounds such as clon_bnds (grid geometry, which a lossy codec would move) unless one is named with --field-to-compress. One pass over the whole field (split across the ranks) finds its value range, its NaN/Inf cells and the time steps and levels on which it varies; the sample is taken from those (a field that is constant on its top levels is not sampled there) and, with --phys-min/--phys-max, reaches the time step and level where the field comes nearest a bound; a field holding NaN/Inf leaves out the codecs that cannot give them back (FixedScaleOffset, zfp, Delta on floats, EBCC). A field whose sample still has no finite value, or a single value while the field varies, says nothing about the codecs: it is skipped as well (an error when it is the --field-to-compress), and a larger --eval-data-size-limit lets the sample reach the variation. A field whose finite values are all one number, or that has none, is stored losslessly with Zstd, without a search and whatever the codec classes. --without-lossy keeps a float field's pipelines bit-exact: compressors and lossless serializers, no filter.
2. compress — persist the fields into one shared .zarr store (dataset opened once) with the winning pipeline of each field, then consolidate the store's metadata so readers open it quickly. A field's manifest_{var}.json must parse and match its sweep_state_{var}.json (a later sweep of the field that has not finished changes it), else the field is not written; with --pipeline the manifest supplies only the gates and chunk geometry. The chunk geometry (--inner-chunk-mib, --max-inner-chunk-mib, --spatial-split) defaults to what the sweep used, as recorded in the manifest, so the store matches what was measured; the command prints where each value came from. The source is read once, in blocks of whole shards; each block is re-read right after its write and the error norms of the whole field are gated against the sweep's thresholds, its physical bounds and cells whose finiteness, or NaN/Inf kind, the round trip changed (--l1-threshold ... --bias-threshold override the thresholds, e.g. for a field without a manifest; the bounds and finiteness gates need none), a FixedScaleOffset pipeline must hold the field's range, and the ratio is compared with the sweep's. Each field is written into a staging store beside the real one ({dataset_stem}.zarr.__staging__) and moved into place only after its gates passed, so an interrupted or failed write never counts as done and never replaces an earlier good array (the next compress deletes what a killed run left in the staging store); a failed field, or one without a usable pipeline, is recorded in batch_manifest.json and makes the command exit with status 1, so a later run retries it. Each array records how it was written (its dc_toolkit attribute, a JSON string: source file, pipeline, chunks, gates, errors, dc_toolkit version); --skip-existing (the default) skips a field only when that record matches what the run would write. Two runs writing one store are refused ({dataset_stem}.zarr.lock).

A combination is identified by its pipeline: the zarr JSON of its three codecs, as stored in zarr.json ({"compressor": {...}, "filter": {...}, "serializer": {...}}, null for an absent codec). It appears in every result row, in the manifests and in the store itself, so nothing has to be rebuilt or re-sampled between the sweep and the write. compress --vars t --pipeline '{...}' (or --pipeline file.json, a manifest_{var}.json included) writes a field with a pipeline of your own, for example one picked from results_{var}.parquet or from a UI.

Output files

evaluate_combos writes the following per variable {var} into --where-to-write:

File	What it is
config_space_{var}.csv	The planned combos in sweep order (name, codec labels, pipeline JSON): the EBCC entries first (the slowest combos), then the valid (compressor, filter, serializer) triples after the pairing rules and --max-evals, shuffled with a count-dependent seed.
config_space_{var}_rank{N}.csv	Per-rank streaming audit trail, one line per combo as it completes, so a killed job loses at most the combos in flight (plus failures_{var}_rank{N}.csv for combos that raised, which the next run retries). Useful to tail during long sweeps, to inspect after a crash, and read back by --resume.
inflight_{var}_rank{N}.csv, crashes_{var}.json	The combo each rank is evaluating, left behind only by a rank that died inside it; the next run counts these in crashes_{var}.json and evaluates them alone.
results_{var}.parquet	Consolidated results across all ranks: one row per combo with its name, codec labels, pipeline JSON, ratio, error metrics, the per-gate verdicts and a keep column marking the combos that passed every gate. The canonical file for analysis (perform_clustering, analyze_clustering), and where compress --stock-codecs-only finds the best stock pipeline.
sweep_state_{var}.json	What the recorded rows were measured on (dataset, sample shape, dtype and digest, sampling policy and vertical floor, chunk settings, physical bounds, a digest of the measuring code, the result-row columns, the definitions of the error metrics, library versions and EBCC's env vars; the full-field value range for information). --resume reuses rows only while it matches; otherwise the field restarts from scratch.
manifest_{var}.json	The best kept combo (best.name, best.pipeline, its ratio, relative L1 error and Euclidean distance), the effective thresholds and physical bounds, the sweep arguments (chunk geometry, codec-space settings, ...), the q99 cut, the counts of rows, kept rows and crashed combos, the environment, the dc_toolkit version and the digest of the sweep state it belongs to. Read by compress and plot_compression_errors.

compress writes the compressed data into {where_to_write}/{dataset_stem}.zarr (the input filename without extension), one zarr array per variable at the root of the store, and batch_manifest.json summarising the run (per field: status, pipeline, ratio, predicted ratio and CR drift, error norms, the verify-gate and CR-drift verdicts, chunk/shard geometry).

HPC parallelism (SLURM / MPI)

How every command parallelizes work (MPI ranks sharing one sample per node, why the production driver runs 32 ranks on a 288-core node, chunks vs shards): docs/PARALLELIZATION.md.

How the sweep samples a field (which time steps and levels, the budget and its memory cap, what a sample cannot see, and the data layouts it does not handle yet): docs/SAMPLING.md.

evaluate_combos runs as one MPI rank per core: the ranks of a node share one copy of the sample through an MPI shared-memory window. The combos still to evaluate are split across the nodes, and inside a node every rank claims the next combo of the node's share from a counter in the node's shared memory each time it finishes one, evaluating one pipeline at a time (the [sweep] line prints the share per node). Scale out by increasing --nodes; the node count is not part of the resume state, so a sweep can be resubmitted on more or fewer nodes:

#SBATCH --nodes=8 --ntasks-per-node=32 --cpus-per-task=1

srun --unbuffered dc_toolkit evaluate_combos input.nc \
    --where-to-write ./out \
    --field-to-compress t \
    --l1-threshold 0.005 \
    --eval-data-size-limit 5GB

A node holds one copy of the sample plus a working set of about twice the sample per rank, about 65 × the sample at 32 ranks; the [memory] line prints the estimate and the sweep shrinks the sample when it does not fit, down to 3 time steps × 3 levels (3 indices across all time-like dims, such as ensemble members; all, where a field has fewer): --eval-data-size-limit is raised to that minimum when it is smaller, and a field whose minimum does not fit stops the sweep with [memcheck] FATAL. docs/PARALLELIZATION.md has the arithmetic. santis.run is the production driver; it reads the input location from the environment and is submitted from the repository root: DYAMOND_DATA_ROOT=/path/to/parent sbatch --account=<account> santis.run, where the parent directory holds the Data_Dyamond_PostProcessed* trees. Each entry of its field list may set its own sample size and ranks per node.

Codec-internal thread pools must be pinned to 1, since every rank, and every dask worker of compress, owns one core (evaluate_combos and compress check this at startup and abort by default; --no-oversubscription-check disables the guard):

export OMP_NUM_THREADS=1 MKL_NUM_THREADS=1 OPENBLAS_NUM_THREADS=1 \
       BLOSC_NTHREADS=1 NUMBA_NUM_THREADS=1 \
       VECLIB_MAXIMUM_THREADS=1 OMP_THREAD_LIMIT=1

compress is a single-process command whose write runs on dask's threaded scheduler: inside a job, give its one task the cores it should use (srun --ntasks=1 --cpus-per-task=32 ...), since --threads (the dask workers and zarr's codec threads) defaults to the cores that task sees and may not exceed them; outside Slurm, invoke it directly. A worker reads a block of whole shards (about 512 MiB), writes it and, with --verify (the default), re-reads it; as many blocks run at once as fit --memory-threshold of the available memory. A netCDF source is read with its chunk cache off, since a block reads a slice of many large HDF5 chunks; its reads are serial (HDF5), so they set the pace. The write is further tuned via --inner-chunk-mib (default: the sweep's value from the manifest, else 16) and --shard-mib (default: 512).

EBCC (optional)

EBCC compresses each (lat, lon) frame with a JPEG 2000 base layer plus an error-bounded residual. At loose error bounds (0.1 to 1 % of the field's range) it reaches 2 to 4x the ratio of zfp; at tight bounds the advantage disappears. It is off by default because it is slow to encode (about 1 to 2 MB/s per core, 200x slower than zfp; decoding is 30 to 200 MB/s) and because of its constraints:

• float fields whose last two dims are a (lat, lon) frame (native ICON grids do not qualify); float64 is down-cast to float32 through the AsType filter;
• no NaN/Inf anywhere in the field: the EBCC library terminates the process on them, so the toolkit checks first and skips or refuses instead;
• one frame (or an exact tile of it, 32 to 2047 cells per side) per inner chunk; --inner-chunk-mib and --spatial-split are ignored, --shard-mib still groups frames into shards;
• runs alone: a filter in front breaks its error bound and a compressor after it gains nothing;
• a store written with EBCC can only be read where dc_toolkit[ebcc] is installed (see Reading a store without dc_toolkit).

Install (needs cmake, a C/C++ toolchain and HDF5 headers; the Docker image includes it):

pip install -e ".[ebcc]"          # or: WITH_EBCC=1 bash install_dc_toolkit.sh

Use --with-ebcc on evaluate_combos to add seven EBCC combos next to the regular sweep, or --serializer-class ebcc to sweep EBCC alone. EBCC is lossy, so both forms need --with-lossy (the default). Each combo bounds the maximum absolute error at a fraction of the field's value range, from 10 % down to 0.01 % (EBCC's own floor is range/65535, its base layer being 16-bit). They appear in the results with no compressor, no filter (or the AsType cast to float32 for float64, which EBCC brings along even when --filter-class excludes it) and are never cut by --max-evals. A winning EBCC pipeline persists like any other (its tile size and error target travel in the pipeline JSON); compress refuses it up front when the field has NaN/Inf or values beyond float32, is not float32 without the AsType filter, or has a frame the tile does not divide. Keep --eval-data-size-limit small on EBCC sweeps: a 5 GB sample takes about an hour per EBCC combo per core.

Reading a store without dc_toolkit

Every codec compress writes decodes in a zarr client without dc_toolkit (given numcodecs, pcodec and zfpy) except two, which exist only through dc_toolkit's zarr.codecs entry point: numcodecs.zfpy_flat (the flattening ZFPY encoder; its bytes are plain zfp, only the name is ours) and numcodecs.ebcc_filter (which also needs the ebcc package). A store holding either fails at zarr.open in a client without them, even for its other arrays, because zarr resolves every array's codec chain when it opens the group's consolidated metadata.

Two ways round it. compress --stock-codecs-only skips such winners and writes the best kept row of results_{var}.parquet whose codecs are all stock, so the store opens anywhere (the sweep still evaluates every pipeline). Or register the name in the reader; for zfpy_flat this needs only zfpy and zarr, no dc_toolkit:

import numcodecs, numcodecs.zfpy
from zarr.codecs.numcodecs import ZFPY
from zarr.registry import register_codec

class _ZFPYFlat(numcodecs.zfpy.ZFPY):          # numcodecs side: the same zfp codec under a second id
    codec_id = "zfpy_flat"
numcodecs.register_codec(_ZFPYFlat)

class ZFPYFlat(ZFPY, codec_name="zfpy_flat"):  # zarr side: the wrapper zarr instantiates from zarr.json
    pass
register_codec("numcodecs.zfpy_flat", ZFPYFlat)

Both registrations are needed: zarr v3 keeps its own codec registry on top of numcodecs'. Decoding needs no reshape logic, because a zfp stream carries its own shape. EBCC has no such shortcut: the reader needs dc_toolkit itself with EBCC (its codec module, dc_toolkit.codecs, needs no MPI); install them from a clone as in Installation, with WITH_EBCC=1 bash install_dc_toolkit.sh.

UI implementation

Two user interfaces wrap the same workflow for one field of a netCDF file: choose the codec space and the relative L1 budget, run evaluate_combos, look at the combinations that passed the gates as KMeans scatter plots of L1 / L2 / LInf vs ratio, pick a pipeline by name, run compress with it and save the store as a zip. The web UI also shows the combinations as a table and exports the plots as HTML; the desktop UI opens the plots in the browser. Outputs go to ./out; the UIs pin the codec thread variables for the commands they launch.

The web UI (streamlit) runs its sweeps under mpirun, one rank per physical core:

dc_toolkit run_web_ui

An upload whose decoded data exceeds 10 MB is cut to a leading block of every dimension so the interactive sweep stays quick; the sweep and the compressed store (<upload>_reduced.zarr.zip) then cover that block only. For the full field run evaluate_combos and compress directly, or use run_web_ui_vcluster, which works on --uploaded_file as it is.

On a vcluster the same UI launches its sweeps under srun with the given allocation and works on a file that is already on the cluster (--uploaded_file, required; --partition defaults to debug); forward the port first (ssh -L 8501:localhost:8501 santis):

dc_toolkit run_web_ui_vcluster \
  --user_account "YOUR_USER_ACCOUNT" \
  --uenv_image "$UENV_NAME" \
  --uploaded_file "PATH_TO_FILE" \
  --time "00:15:00" \
  --nodes "1" --ntasks-per-node "32"

evaluate_combos runs one MPI rank per core; --ntasks-per-node is the number of ranks, and cores, per node.

The desktop UI (Qt; installs PyQt6 on first use) runs locally:

dc_toolkit run_local_ui

Docker

The Dockerfile builds a self-contained image (all dependencies, and the repository's default branch cloned from GitHub: local changes are not in the image):

docker build -t dc-toolkit .

An example run:

docker run \
  -u $(id -u):$(id -g) \
  -w /mnt/data/docker_saved_files \
  -v "$(pwd)/netCDF_files":/mnt/data \
  -e XDG_CACHE_HOME=/tmp/.cache \
  -e OMP_NUM_THREADS=1 -e MKL_NUM_THREADS=1 -e OPENBLAS_NUM_THREADS=1 \
  -e BLOSC_NTHREADS=1 -e NUMBA_NUM_THREADS=1 \
  -e VECLIB_MAXIMUM_THREADS=1 -e OMP_THREAD_LIMIT=1 \
  --entrypoint /bin/bash \
  dc-toolkit \
  -c 'dc_toolkit evaluate_combos /opt/data-compression/netCDF_files/tigge_pl_t_q_dx=2_2024_08_02.nc --where-to-write /mnt/data/docker_saved_files --field-to-compress t --l1-threshold 0.005'

Command Breakdown:

• -u $(id -u):$(id -g): Runs the container using your local machine's User and Group IDs rather than the Docker default root. The files written to your machine are then owned by you and aren't locked behind root permissions.
• -w /mnt/data/docker_saved_files: Sets the Working Directory.
• -v "$(pwd)/netCDF_files":/mnt/data: The volume mount. This creates a bridge between your local computer and the container so the toolkit can read your input data and write the results back to your hard drive.
• -e XDG_CACHE_HOME=/tmp/.cache: Sets the cache directory to a temporary location inside the container.
• -e OMP_NUM_THREADS=1 ...: Pins the codec-internal thread pools to 1; evaluate_combos aborts at startup otherwise (--no-oversubscription-check disables the guard).
• --entrypoint /bin/bash: Forces Docker to start with a Bash shell instead of the default program (dc_toolkit).
• dc-toolkit: The name of the Docker image to run.
• -c '...': The shell command the container runs:
  • dc_toolkit evaluate_combos ...: Executes the actual compression tool with a single rank (the next section runs one rank per core), using a file inside the container and saving the results (under --where-to-write) to your mounted volume.

Or for the web UI:

docker run -p 8501:8501 dc-toolkit run_web_ui

Running with MPI (single-container, exercises the MPI code path)

OpenMPI + Docker requires specific file permission and cache handling. On a single container evaluate_combos runs one MPI rank per core the container may use (-n 4 below); the ranks share one copy of the sample. Open MPI keeps that copy in the container's /dev/shm, which Docker limits to 64 MB: give it at least the sample size with --shm-size (5g covers the default --eval-data-size-limit), or a sweep of a real field on several ranks aborts with [shared-sample] FATAL.

---

Mac and Linux

docker run \
  -u $(id -u):$(id -g) \
  -w /mnt/data/docker_saved_files \
  -v $(pwd)/netCDF_files:/mnt/data \
  --shm-size=5g \
  -e OMP_NUM_THREADS=1 -e MKL_NUM_THREADS=1 -e OPENBLAS_NUM_THREADS=1 \
  -e BLOSC_NTHREADS=1 -e NUMBA_NUM_THREADS=1 \
  -e VECLIB_MAXIMUM_THREADS=1 -e OMP_THREAD_LIMIT=1 \
  --entrypoint mpirun \
  dc-toolkit \
  -n 4 \
  bash -c 'HOME=/tmp/$OMPI_COMM_WORLD_RANK exec dc_toolkit evaluate_combos /opt/data-compression/netCDF_files/tigge_pl_t_q_dx=2_2024_08_02.nc --where-to-write /mnt/data/docker_saved_files --field-to-compress t --l1-threshold 0.005 --eval-data-size-limit 5GB'

Command Breakdown:

• -u $(id -u):$(id -g): Runs the container as your local user so outputs aren't locked behind root permissions.
• -w /mnt/data/docker_saved_files: Sets the Working Directory.
• -v $(pwd)/netCDF_files:/mnt/data: Volume mount bridging local and container filesystems.
• --shm-size=5g: Room in /dev/shm for the sample the ranks share; at least the sample size.
• -e OMP_NUM_THREADS=1 ...: Pins codec-internal thread pools to 1: every rank owns one core.
• --entrypoint mpirun: Replaces the default entrypoint (dc_toolkit) with OpenMPI's launcher.
• dc-toolkit: The image name.
• -n 4: Four MPI ranks, one per core the container may use; they share one copy of the sample.
• bash -c '...': Executes the dc_toolkit command:
  • HOME=/tmp/$OMPI_COMM_WORLD_RANK: A $HOME per rank under the writable /tmp (/tmp/0, /tmp/1, ...), so ranks do not share caches.
  • exec dc_toolkit evaluate_combos ... --where-to-write /mnt/data/docker_saved_files ...: Runs the sweep, writing all outputs into the mounted volume.

---

Windows (PowerShell)

When using Docker Desktop on Windows via WSL 2, Docker handles file permissions differently. You don't need to pass your user ID (Docker Desktop handles the translation automatically), but you do need to explicitly allow OpenMPI to run as root and format your paths for PowerShell.

docker run `
  -e HOME=/tmp `
  -e OMP_NUM_THREADS=1 -e MKL_NUM_THREADS=1 -e OPENBLAS_NUM_THREADS=1 `
  -e BLOSC_NTHREADS=1 -e NUMBA_NUM_THREADS=1 `
  -e VECLIB_MAXIMUM_THREADS=1 -e OMP_THREAD_LIMIT=1 `
  -w /mnt/data/docker_saved_files `
  -v "${PWD}\netCDF_files:/mnt/data" `
  --shm-size=5g `
  --entrypoint mpirun `
  dc-toolkit `
  --allow-run-as-root `
  -n 4 `
  bash -c "HOME=/tmp/`$OMPI_COMM_WORLD_RANK exec dc_toolkit evaluate_combos /mnt/data/tigge_pl_t_q_dx=2_2024_08_02.nc --where-to-write /mnt/data/docker_saved_files --field-to-compress t --l1-threshold 0.005 --eval-data-size-limit 5GB"

Command Breakdown:

• -e HOME=/tmp: Sets a base temporary home directory for the container environment.
• -e OMP_NUM_THREADS=1 ...: Pins codec-internal thread pools to 1 (prevents nested oversubscription).
• -w /mnt/data/docker_saved_files: Sets the Working Directory inside the container.
• -v "${PWD}\netCDF_files:/mnt/data": Windows equivalent of the volume mount. ${PWD} dynamically grabs your current PowerShell directory to link your local files to the container.
• --shm-size=5g: Room in /dev/shm for the sample the ranks share; at least the sample size.
• --entrypoint mpirun: Replaces the default container start command with OpenMPI's launcher.
• dc-toolkit: The image name.
• --allow-run-as-root: The container defaults to root on Windows; this flag lifts OpenMPI's built-in refusal to run parallel jobs as root.
• -n 4: Four MPI ranks, one per core the container may use.
• bash -c "...": Executes the parallel command. Note the double quotes for PowerShell, and PowerShell's escape character, a backtick, in front of the MPI variable (`$OMPI_COMM_WORLD_RANK) to prevent PowerShell from evaluating it on your host before it reaches the container.

Slides

Click here to view slides