thermograph/infra/lake-iceberg/README.md

lake-iceberg — Iceberg conversion for the ERA5 lake

Maintains an Apache Iceberg v2 table era5_daily at s3://era5-thermograph/iceberg/era5_daily over the lake's existing hive parquet (era5/daily/tile=<ti>_<tj>/year=<Y>/month=<M>/part.parquet).

Key properties:

• No data rewrite. sync_iceberg.py uses pyiceberg add_files: the Iceberg metadata references the existing part files in place. Storage is not doubled and nothing under era5/ is ever written, moved, or deleted — the tool only writes under the iceberg/ prefix.
• Incremental + idempotent. Each run diffs the tiles in era5/manifest.parquet (the source of truth for complete tiles — the extractor uploads all of a tile's objects before its manifest rows appear) against the thermograph.synced-tiles table property, and registers only the new ones. Files and the property update land in the same commit, so an interrupted run never half-syncs a tile; re-running is always safe, even while an extraction job is writing new tiles.
• Self-contained catalog. A pyiceberg SqlCatalog on local sqlite (THERMOGRAPH_ICEBERG_CATALOG_DB, default /state/iceberg-catalog.db). The catalog is only a pointer: if the db is lost, the next run finds the newest *.metadata.json under the table location and re-registers it — the synced-tiles set lives in the table metadata itself, so nothing is re-added.
• Partitioning (tile, year, month) via order-preserving transforms of real columns: truncate[12](lat_idx) / truncate[12](lon_idx) (partition values are ti*12 / tj*12 — the part files carry no tile column, the 12 matches the lake's 12x12-point tile grid) plus year(date) and month(date). Predicates on lat_idx/lon_idx/date prune partitions directly — no synthetic tile/year/month columns needed in queries.

Build

docker build -t thermograph-lake-iceberg infra/lake-iceberg/

Run (one-shot)

Config is the backend's lake variables (same names and defaults as backend/data/era5lake.py): THERMOGRAPH_LAKE_S3_ACCESS_KEY, THERMOGRAPH_LAKE_S3_SECRET_KEY, and optionally THERMOGRAPH_LAKE_S3_ENDPOINT / _BUCKET / _REGION.

docker run --rm --env-file /path/to/lake.env \
    -v lake-iceberg-state:/state \
    thermograph-lake-iceberg

Flags: --dry-run (list pending tiles, write nothing), --limit N (cap tiles per run), --batch N (tiles per Iceberg commit, default 5), --catalog-db PATH. PYICEBERG_MAX_WORKERS controls the parquet-footer read parallelism inside add_files (32 is a good value for the ~1000 small footers per tile).

Each run prints the row-count reconciliation (records added vs the manifest's per-tile row counts) and ends with the table's current metadata JSON path. It also refreshes metadata/version-hint.text (Hadoop-catalog convention) so readers can scan the bare table root without knowing that path. Transient S3 errors (Contabo intermittently returns SLOW_DOWN / connection timeouts under load) are retried per batch with backoff (--retries, default 3); a run that still dies resumes at the next batch boundary on re-run.

Scheduling

Meant to run as a one-shot on prod after extraction batches finish — e.g. a cron or an ops-cron workflow step invoking the docker run above, with the secrets rendered from the SOPS vault like every other consumer of the THERMOGRAPH_LAKE_S3_* pair. Deliberately not wired into the Swarm stack or CI: the sync is an offline maintenance job, not a service, and each run costs one footer read per new part file (~1000 per tile), so it belongs after extraction batches, not on a tight loop.

Querying from DuckDB

No catalog server needed — the version-hint pointer lets DuckDB scan the table root directly:

INSTALL iceberg; LOAD iceberg;
INSTALL httpfs;  LOAD httpfs;
CREATE SECRET lake (
    TYPE s3,
    KEY_ID '<THERMOGRAPH_LAKE_S3_ACCESS_KEY>',
    SECRET '<THERMOGRAPH_LAKE_S3_SECRET_KEY>',
    ENDPOINT 'eu2.contabostorage.com',
    REGION 'default',
    URL_STYLE 'path'
);
SELECT count(*)
FROM iceberg_scan('s3://era5-thermograph/iceberg/era5_daily');

To pin an exact table version, scan the metadata JSON path the sync run printed (iceberg_scan('s3://…/metadata/<NNNNN-uuid>.metadata.json')).

Partition pruning works on the raw columns, e.g. one tile-month:

SELECT lat_idx, lon_idx, avg(tmax)
FROM iceberg_scan('s3://era5-thermograph/iceberg/era5_daily')
WHERE lat_idx BETWEEN 120 AND 131 AND lon_idx BETWEEN 120 AND 131
  AND date >= DATE '2020-07-01' AND date < DATE '2020-08-01'
GROUP BY 1, 2;

The part files carry no Iceberg field ids; the table sets schema.name-mapping.default so engines resolve columns by name (validated with pyiceberg 0.11.1 and DuckDB 1.5.5, values byte-identical to the hive side).

Tests

pip install -r infra/lake-iceberg/requirements.txt pytest
pytest infra/lake-iceberg/test_sync.py

Local-filesystem fixtures only (tiny parquet files, file:// warehouse, sqlite catalog in tmp) — no network, no credentials.