Python Maintain a Collection

This recipe keeps a collection of NetCDF files in one table. New files arrive each day, and some files get corrections.

The steps

1. Add the new files with one writer. Give each dataset a source_file attribute.
2. Replace the corrected files: delete their datasets, then add the new datasets.
3. Compact the table.
4. Free the disk space of old versions, with a margin for readers.

The script

import logging
from datetime import timedelta
from pathlib import Path

import xarray as xr
import lance_nd

URI = "profiles.lance"


def load(path):
    with xr.open_dataset(path) as ds:
        ds = ds.load()
    ds.attrs["source_file"] = path.name
    return ds


def quoted(values):
    return ", ".join("'" + v.replace("'", "''") + "'" for v in values)


def add(paths):
    """Add the datasets of `paths` in one commit. Skip files that fail."""
    with lance_nd.append(URI) as writer:
        for path in paths:
            try:
                writer.write(load(path))
            except (OSError, lance_nd.LanceNdError) as error:
                logging.warning("skipped %s: %s", path, error)


new_files = sorted(Path("incoming/new").glob("*.nc"))
corrected = sorted(Path("incoming/corrected").glob("*.nc"))

# 1. New files.
add(new_files)

# 2. Corrected files: delete the old datasets, then add the new datasets.
table = lance_nd.open(URI)
if corrected:
    table.delete(f"__global_attrs__.source_file IN ({quoted(p.name for p in corrected)})")
    add(corrected)

# 3. Compact.
table = lance_nd.open(URI)
table.compact(target_fragment_bytes=512 << 20)

# 4. Free disk space. Keep one day of versions for readers.
table.cleanup_old_versions(older_than=timedelta(days=1))

Notes

• One commit for many files. add() uses one writer for all files of the day, so the table gets one version and full fragments.
• Failed files. A file that cannot be opened or that does not fit the table is logged and skipped. The other files are committed.
• Many files at once. For tens of thousands of files, commit each few thousand files: open a new writer for each batch. A failed batch then loses less work.
• Fast reads with workers. Opening NetCDF files is often slower than the write. Read the files in a concurrent.futures.ProcessPoolExecutor, and write the datasets in the main process.
• Stats. Create the table with stats=True, so that region and value queries can skip datasets.
• Margin. A reader that opened an older version fails when its files are removed. One day is enough for most readers.

Create the table once

with lance_nd.create(URI, stats=True, column_defaults={"compression": "zstd"}) as writer:
    for path in sorted(Path("archive").glob("*.nc")):
        writer.write(load(path))

See Create a table for the options.