Iceberg conversion container for the ERA5 lake (infra/lake-iceberg) #24

Merged

admin_emi merged 1 commit from feat/lake-iceberg into dev

2026-07-23 22:56:29 +00:00

admin_emi commented

2026-07-23 22:50:06 +00:00

Owner

What

infra/lake-iceberg/: a self-contained one-shot container that 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).

sync_iceberg.py — incremental sync: diffs the tiles in era5/manifest.parquet against the thermograph.synced-tiles table property and registers only the new ones. The property updates in the same Iceberg commit as the files, so runs are idempotent, resume at batch boundaries, and are safe while the extraction job keeps writing (the manifest only lists tiles whose objects are all uploaded — it is the completeness signal, not bucket listing). Transient S3 errors retry per batch with backoff. Every run reconciles added records against the manifest's per-tile row counts, refreshes metadata/version-hint.text, and prints the current metadata JSON path.
Config via the backend's THERMOGRAPH_LAKE_S3_* names/defaults (backend/data/era5lake.py); the catalog is a local sqlite SqlCatalog (THERMOGRAPH_ICEBERG_CATALOG_DB, default /state/iceberg-catalog.db) that self-heals by re-registering the latest metadata JSON if the db is lost.
Strictly additive to the bucket: all writes land under iceberg/; nothing under era5/ is written, moved, or deleted.
Dockerfile (python:3.12-slim; pinned pyiceberg 0.11.1 / pyarrow 25.0.0 / s3fs 2026.6.0), README.md, test_sync.py (11 pytest cases on local-filesystem fixtures — tile diffing, add_files planning, pruning, catalog-loss recovery, retries; no network).

add_files, not a rewrite

The table registers the existing part files in place (pyiceberg add_files): no data movement, no storage doubling. The part files carry no tile/year/month columns (they are path components), so identity partitioning would have forced a rewrite; instead the spec uses order-preserving transforms of real columns — truncate[12](lat_idx), truncate[12](lon_idx) (values ti*12/tj*12, i.e. the tile), year(date), month(date). Each part file holds exactly one tile-month, so add_files derives every partition value from footer min/max stats alone. Queries prune on plain lat_idx/lon_idx/date predicates, and schema.name-mapping.default is set so engines resolve the id-less parquet columns by name.

End-to-end proof (real bucket, extraction still running)

At proof time the manifest listed 55 tiles and the extraction job was still adding (~1 tile / 75s); 36 tiles were synced and verified — catching the remainder is exactly the container's designed usage: re-run it after extraction batches.

Synced: 36 tiles → 37,260 data files, 163,306,368 records; manifest rows for that tile set: 163,306,368 (every batch reconciled ok).
DuckDB 1.5.5: SELECT count(*) FROM iceberg_scan('s3://era5-thermograph/iceberg/era5_daily') → 163,306,368 (4.9s, bare table root via version-hint); identical via the explicit metadata path (0.9s).
Metadata at proof time: s3://era5-thermograph/iceberg/era5_daily/metadata/00008-5138938f-aa5f-4cb6-9b67-75183eb1d074.metadata.json
Spot probe (tile 11_18, 2015-01): count / min(date) / max(date) / avg(tmax) / avg(feels) / count(sun) identical between iceberg_scan and read_parquet of the hive part file (4464 rows; avg tmax 15.9916; sun all-null).
docker build of infra/lake-iceberg/ verified locally; the image runs --help and exits cleanly when unconfigured.

Notes

Contabo path-style S3: pyiceberg PyArrowFileIO needs s3.endpoint + s3.force-virtual-addressing=false; s3fs addressing_style: path; DuckDB secret URL_STYLE 'path'. With those set, everything (including REGION 'default') worked without further quirks on pyiceberg 0.11.1.
Listing uses one flat find() per tile: fsspec glob over year=*/month=* walks every directory level (~1000 LISTs ≈ 3 min per tile on this endpoint).
Contabo intermittently returns SLOW_DOWN / slow-transfer timeouts under load; the per-batch retry absorbed real occurrences during the proof runs.
Swarm stack / CI wiring is deliberately out of scope for this PR: the sync is an offline maintenance job (cron / one-shot on prod after extraction batches), per the README.

## What `infra/lake-iceberg/`: a self-contained one-shot container that 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`). - `sync_iceberg.py` — incremental sync: diffs the tiles in `era5/manifest.parquet` against the `thermograph.synced-tiles` table property and registers only the new ones. The property updates in the same Iceberg commit as the files, so runs are idempotent, resume at batch boundaries, and are safe while the extraction job keeps writing (the manifest only lists tiles whose objects are all uploaded — it is the completeness signal, not bucket listing). Transient S3 errors retry per batch with backoff. Every run reconciles added records against the manifest's per-tile row counts, refreshes `metadata/version-hint.text`, and prints the current metadata JSON path. - Config via the backend's `THERMOGRAPH_LAKE_S3_*` names/defaults (`backend/data/era5lake.py`); the catalog is a local sqlite SqlCatalog (`THERMOGRAPH_ICEBERG_CATALOG_DB`, default `/state/iceberg-catalog.db`) that self-heals by re-registering the latest metadata JSON if the db is lost. - Strictly additive to the bucket: all writes land under `iceberg/`; nothing under `era5/` is written, moved, or deleted. - `Dockerfile` (python:3.12-slim; pinned pyiceberg 0.11.1 / pyarrow 25.0.0 / s3fs 2026.6.0), `README.md`, `test_sync.py` (11 pytest cases on local-filesystem fixtures — tile diffing, add_files planning, pruning, catalog-loss recovery, retries; no network). ## add_files, not a rewrite The table registers the existing part files in place (pyiceberg `add_files`): no data movement, no storage doubling. The part files carry no tile/year/month columns (they are path components), so identity partitioning would have forced a rewrite; instead the spec uses order-preserving transforms of real columns — `truncate[12](lat_idx)`, `truncate[12](lon_idx)` (values `ti*12`/`tj*12`, i.e. the tile), `year(date)`, `month(date)`. Each part file holds exactly one tile-month, so `add_files` derives every partition value from footer min/max stats alone. Queries prune on plain `lat_idx`/`lon_idx`/`date` predicates, and `schema.name-mapping.default` is set so engines resolve the id-less parquet columns by name. ## End-to-end proof (real bucket, extraction still running) At proof time the manifest listed 55 tiles and the extraction job was still adding (~1 tile / 75s); 36 tiles were synced and verified — catching the remainder is exactly the container's designed usage: re-run it after extraction batches. - Synced: 36 tiles → 37,260 data files, 163,306,368 records; manifest rows for that tile set: 163,306,368 (every batch reconciled `ok`). - DuckDB 1.5.5: `SELECT count(*) FROM iceberg_scan('s3://era5-thermograph/iceberg/era5_daily')` → 163,306,368 (4.9s, bare table root via version-hint); identical via the explicit metadata path (0.9s). - Metadata at proof time: `s3://era5-thermograph/iceberg/era5_daily/metadata/00008-5138938f-aa5f-4cb6-9b67-75183eb1d074.metadata.json` - Spot probe (tile 11_18, 2015-01): count / min(date) / max(date) / avg(tmax) / avg(feels) / count(sun) identical between `iceberg_scan` and `read_parquet` of the hive part file (4464 rows; avg tmax 15.9916; sun all-null). - `docker build` of `infra/lake-iceberg/` verified locally; the image runs `--help` and exits cleanly when unconfigured. ## Notes - Contabo path-style S3: pyiceberg PyArrowFileIO needs `s3.endpoint` + `s3.force-virtual-addressing=false`; s3fs `addressing_style: path`; DuckDB secret `URL_STYLE 'path'`. With those set, everything (including `REGION 'default'`) worked without further quirks on pyiceberg 0.11.1. - Listing uses one flat `find()` per tile: fsspec `glob` over `year=*/month=*` walks every directory level (~1000 LISTs ≈ 3 min per tile on this endpoint). - Contabo intermittently returns SLOW_DOWN / slow-transfer timeouts under load; the per-batch retry absorbed real occurrences during the proof runs. - Swarm stack / CI wiring is deliberately out of scope for this PR: the sync is an offline maintenance job (cron / one-shot on prod after extraction batches), per the README.

admin_emi added 1 commit 2026-07-23 22:50:08 +00:00

Add infra/lake-iceberg: Iceberg conversion for the ERA5 lake

PR build (required check) / changes (pull_request) Successful in 9s

Details

secrets-guard / encrypted (pull_request) Successful in 6s

Details

PR build (required check) / build-frontend (pull_request) Has been skipped

Details

PR build (required check) / validate-observability (pull_request) Has been skipped

Details

PR build (required check) / build-backend (pull_request) Successful in 44s

Details

PR build (required check) / gate (pull_request) Successful in 2s

Details

f87e22be51

Registers the hive part files under era5/daily into an Apache Iceberg v2
table at iceberg/era5_daily via pyiceberg add_files -- the metadata points
at the existing parquet in place, no data rewrite. Incremental: each run
diffs era5/manifest.parquet (the completeness signal while extraction is
running) against the thermograph.synced-tiles table property, updated in
the same commit as the files, so runs are idempotent and resume at batch
boundaries. Partitioned by truncate[12](lat_idx), truncate[12](lon_idx),
year(date), month(date) -- order-preserving transforms add_files derives
from footer stats; queries prune on plain column predicates. Local sqlite
catalog that re-registers from the latest metadata JSON if lost;
version-hint.text refreshed per run so DuckDB can iceberg_scan the bare
table root. Per-batch retries with backoff for Contabo SLOW_DOWN and
slow-transfer timeouts. Slim pinned one-shot image; local-filesystem
pytest suite (no network).