data-apache-lakehouse

Files (3)

• SKILL.md
• pyiceberg-capabilities.md · references/
• single-host-operations.md · references/

SKILL.md (first lines)

# Apache Iceberg Lakehouse

Domain skill for building and operating Python-first Apache Iceberg lakehouses: PyIceberg + Polars/DuckDB medallion writes, REST/Glue catalog choice, and the architectural discipline that keeps incremental jobs incremental.

> **Verified:** 2026-05-28 against PyIceberg 0.11.1, pyiceberg-core 0.8.0, Iceberg spec V2/V3.
> Claims tagged `[v0.11]` and the **PyIceberg capability matrix** (in
> [`references/pyiceberg-capabilities.md`](references/pyiceberg-capabilities.md)) are
> version-sensitive — re-check against your installed version (`pip show pyiceberg`) before trusting
> them; the matrix lists how to verify each in seconds. Everything else (Iceberg internals, the
> architectural principles) is format/architecture-level and ages slowly.

> **Part of the `data` skill family.** The cross-cutting pipeline principles (idempotency, watermarks, schema fencing, resilience, bounded memory) are stated generally in the **data** hub; this skill is their Apache Iceberg expression. For consuming or serving APIs around the lakehouse, see **data-api**.

## When to invoke this skill

- Designing a new pipeline, layer, or transform in a bronze/silver/gold lakehouse.
- Adding or modifying a gold aggregate, analytics view, or derived table.
- Debugging incremental jobs that get slower as data grows, scheduled jobs that overrun their window, lock contention, memory pressure, or "metadata bigger than data."
- Reviewing any PR that touches a layer boundary (bronze→silver, silver→gold) or the catalog config.
- Choosing or swapping a catalog backend (Glue, REST, Polaris, Lakekeeper, Nessie, JDBC, Hive).
- Picking compaction / expire-snapshots / orphan-file-removal cadence for a table.
- Deciding between Iceberg, Delta, and Hudi for a new lakehouse.
- Investigating a bad publish — finding the right snapshot to roll back to, auditing what a backfill landed.

If you're about to add a global lock, a "rebuild from full history" path, a Parquet mirror next to an Iceberg table, a cron that recomputes a snapshot from scratch, or a write that bypasses the catalog — stop and read this skill first.

## Medallion rules

1. **Bronze** — raw ingest, minimal transforms, append-friendly.
2. **Silver** — typed, deduped, schema-validated; fail closed on drift when correctness depends on it.
3. **Gold** — business aggregates; read silver/bronze only; no back-writing to lower layers in the same job.

## Iceberg internals worth knowing (the parts that change how you write code)

The skill assumes you know the layered model (datafiles → manifests → manifest list → metadata.json → catalog). These are the facts that actually change PyIceberg-writer behavior:

- **Pruning happens twice, at different layers.** Manifest entries carry per-column lower/upper bounds, evaluated at *planning* time (cheap Avro reads, zero datafiles touched). Parquet row groups carry their own min/max, evaluated at *scan* time after the file is opened. A predicate on a sorted/partitioned column gets both layers; a predicate on an unsorted high-cardinality column gets ~no manifest pruning. **This is why sort order matters: it tightens manifest bounds.**
- **Manifest lists carry partition bounds and per-snapshot sequence numbers.** Diffing two snapshots is a manifest-list-only walk — you never need to open datafiles to find what changed. This is the underlying machinery of incremental reads even when PyIceberg only exposes it indirectly via watermark columns.
- **`last-sequence-number` in `metadata.json` is the OCC primitive.** Every commit bumps it. Equality deletes are scoped by it (a delete with sequence N only applies to data files with sequence < N). A re-inserted row gets a higher sequence number and survives. This is also what `CommitFailedException` retries depend on — your exactly-once primitive in PyIceberg.
- **The catalog stores exactly one pointer per table** — the current `metadata.json` path. Atomic CAS on that pointer is the entire basis for ACID. Hadoop catalog on S3 has no atomic rename → corruption under concurrent writers. Use REST/Glue/Nessie/JDBC for any multi-writer S3 deployment.
- **Manifests are Avro, never mixed.** A manifest tracks datafiles or delete files, never both. Reading manifests is fast; rewriting them (`rewrite_manifests`) is cheap relative to data rewrites.
- **Stats are written at commit time, not by a separate job.** Unlike Hive, Iceberg manifests are always fresh because writers compute bounds inline. Don't add a post-write "stats job" — it doesn't exist.
- **Puffin files** carry precomputed sketches (currently Theta sketches for approximate distinct counts). PyIceberg does not read or write them today. Don't pin a `COUNT DISTINCT` SLA to puffin if PyIceberg is in the loop.

## Architectural principles

Load-bearing properties of any Iceberg lakehouse transform. Each principle is paired with a falsifiable test — if you can't pass the test, the principle is being violated, not "applied differently."

### Forward motion is a watermark, not a scan

Every layer-to-layer promote carries a persisted high-water mark on its time/sequence column (`event_ts`, `sequence_id`, `updated_at`, etc.). The watermark commit is part of the same atomic write as the data — not a side-effect afterward. A transform with no watermark is a global recompute waiting to happen; it passes tests today and gets slower in proportion to data growth, with no warning. "Rebuild from history" exists only behind an explicit `--rebuild` flag, never as the default code path.

PyIceberg has no `incrementalAppendScan` yet (Java does). A watermark column committed atomically with the data is the substitute, and it lines up with Iceberg's own model: each commit bumps `last-sequence-number`; your watermark carries equivalent semantics at the row level.

**Concrete implementation — the watermark is a table property committed *with* the data.** Store the last-consumed source position (max `event_ts`/`sequence_id`, or the source snapshot id) as a property on the *target* table, e.g. `wm.<source>.<kind>`. The clean form is one transaction — `with target.transaction() as tx: tx.append(batch); tx.set_properties({"wm.sales.silver": str(new_wm)})` — so data and watermark land in a single atomic commit and can never disagree. A two-commit form (data, then a separate property bump) is acceptable **only** when every write is an idempotent upsert/partition-overwrite, so a crash between the two commits merely re-processes the last window. It's metadata-only, free to read at planning time, and needs no extra sidecar table. Full recipe: [`references/single-host-operations.md`](references/single-host-operations.md).

**Test:** if the upstream table grew 10× tomorrow, would this transform's runtime grow 10×? If yes, it lacks a watermark.

### A watermarked entity delta is still a windowed scan

Selecting *keys* incrementally (`entities with ts > watermark`) then reading `WHERE entity_id IN (keys)` **without** a time predicate is not incremental. Iceberg will not prune `month(ts)` partitions; PyIceberg COW then rewrites every data file that contains those ids — often a whole month file to flip a few flag columns. That is the entity-scoped O(history) leak from the **data** hub.

Fix on the timer path:

1. Predicate **both** keys and time: `entity_id IN (...) AND ts >= now - lookback` (lookback on the **partition** column, so months drop at planning).
2. Overwrite **only the sale_ids / rows in that window**. Do not emit 2016 rows into the COW filter because you scored a 2026 sale.
3. `--full` / `--rebuild` is the lifetime path. Do not make it the default.
4. Layout (sort by entity_id inside the time partition) tightens file bounds so `IN (entity)` skips files. It is not a substitute for the time predicate. Iceberg has no B-tree index.

**Test:** one new row today for an entity that has 10 years of history — does `plan_files()` for the score/overwrite include 2016 partitions? If yes, the window is missing.

### A `latest` / `as_of` pointer is a global tip, not a per-chunk side effect

A file or table property named `latest`, `as_of`, `tip`, or `watermark` that readers use for
`latest_as_of()` is the **global tip**. Write it once after a multi-day / multi-partition loop
succeeds. Updating it inside the loop leaves the oldest (or last-iterated) day in the pointer.
Backfill helpers default `update_latest=False`; the daily product pass sets `True` only after
today's write. Per-day overwrite can be idempotent and still be tip-unsafe.

**Test:** backfill N days with a pre-seeded today pointer. Is the tip still today / max day, or
min day? If min day, the pointer was written inside the loop.

### Every aggregate has a declared shape

Each gold/analytics aggregate is one of three shapes, and the shape is named in the module:

- **Append-only ledger** — immutable primary key; `append` on the delta dedupes by key.
- **Point-update keyed snapshot** — `MERGE INTO` on the keys touched by the delta. Cost is O(delta), not O(history).
- **Windowed scan** — bounded `source.filter(ts > now - window)` with partition pruning, on its own cadence.

"Full rebuild" is none of these. If you can't put one of the three labels on an aggregate, the architecture is wrong — don't slow the cadence to mask the cost, reshape the aggregate.

**Test:** can you write `# shape: append | point_update | windowed_scan` at the top of the aggregate's module without lying? If no, redesign.

### Maintenance is owned, not hoped for

PyIceberg can now **expire snapshots** `[v0.11]` (`table.maintenance.expire_snapshots()`), but this is **metadata-only — it does not physically delete the orphaned data files** (see the capability matrix). It cannot compact or remove orphan files at all. Physical file GC and compaction are Spark procedures, Trino `ALTER TABLE ... EXECUTE`, or a managed service. Untended Iceberg tables silently grow until metadata reads dominate query time and small files defeat manifest pruning. The safe order is fixed: `rewrite_data_files` → `expire_snapshots` (≥3-day window) → `remove_orphan_files` (≥3-day window) → `rewrite_manifests`. Skipping the safety window is how you delete files a still-running query needs.

There are now three legitimate ways to own maintenance — pick one explicitly, don't leave it implicit:

1. **External engine** — Spark procedures or Trino `EXECUTE optimize` on a schedule. Most control; you operate the cluster.
2. **Managed / zero-ops storage** `[2025]` — the catalog or storage layer runs policy-driven compaction, snapshot expiration, **and** orphan-file removal on the backend, decoupled from your Python pipeline. Examples: Amazon S3 Tables (continuous compaction + expiration since 2025), Cloudflare R2 Data Catalog, Polaris-managed maintenance. This is the only option where expiry *and* physical GC both happen without you wiring a second job.
3. **Hybrid** — PyIceberg `expire_snapshots` for metadata hygiene from your pipeline, plus a scheduled external/managed job for the physical `remove_orphan_files` + compaction it cannot do.

On a single host with no cluster, the "external engine" is usually a `systemd` timer (or cron) that runs PyIceberg `expire_snapshots` plus a scheduled DuckDB/Spark compaction + orphan-removal step — a legitimate option 1, just hand-wired (see [`references/single-host-operations.md`](references/single-host-operations.md)).

**Test:** name the person/service, the cadence, and the command (or backend policy) that maintains this table — **and** confirm physical file GC is covered, not just metadata expiry. If any of those is "I'll figure it out," the maintenance does not exist.

### Validation belongs on a branch, not in production

PyIceberg supports branches and tags since 0.8. Treat the `main` branch of an Iceberg table as published state; do risky work (large backfills, schema reshuffles, suspicious silver→gold rewrites) on a named branch, validate against it, then merge or discard. This is the write-audit-publish pattern expressed natively in Iceberg. A staging mirror that lives outside the catalog is a hidden interface; an Iceberg branch is a first-class one. Nessie adds multi-table atomic branch merges if you need them.

**Test:** if this promotion failed audit, could you discard it by dropping a branch? If the answer involves manually rewinding writes to the live table, it shouldn't have been on the live table.

### Schema evolution is an operator action

Additive columns, type widenings, and partition-spec changes are not a silent writer side effect.
Evolve on a branch (or a dedicated evolve script), then readers. PyIceberg: `delete_column`, not
`remove_field` (the latter orphans the field in manifests). Revert-then-readd of the same name is

From Evan-Kim2028/agent-skills, MIT.

More from Evan-Kim2028/agent-skills