Lab 10 — Create an Iceberg Table, Write Snapshots, Time-Travel Between Them

Lesson 10 · Data Engineering with Python · ~45 min

By the end of this lab you will have: installed pyiceberg and DuckDB’s iceberg extension; created the default.events Iceberg table via the sample-pipeline code; appended the L02 Parquet partitions twice (two snapshots); queried each snapshot from DuckDB and confirmed different row counts; evolved the table schema; and run expire_snapshots to observe metadata pruning.

Prerequisites

• L09 lab complete — DuckDB is installed and the L02 Parquet partitions exist at data-engineering/sample-pipeline/data/parquet/date=*/.
• Python 3.10+ virtual environment with the course requirements.txt installed (covers pyiceberg[sql-sqlite,pyarrow,duckdb]>=0.7).
• One terminal you can run python -m … in from data-engineering/sample-pipeline/.

Step 1 — Install dependencies and inspect the table definition

Install the Module 4 dependencies if you have not already:

pip install -r data-engineering/sample-pipeline/requirements.txt
python -c "import pyiceberg; print(pyiceberg.__version__)"

Expected: a version string >=0.7.

Now open data-engineering/sample-pipeline/iceberg/tables.py and walk through the key definitions:

• SCHEMA — six NestedField entries, each with a stable numeric field_id (1 through 6). Field IDs never change even if you rename a column; that is how Iceberg preserves backwards compatibility across snapshots.
• The column for the event actor is named actor (field ID 3) — this is the canonical name in the Iceberg schema. The source L02 Parquet also uses actor, so no renaming is needed on the Iceberg side for that column.
• PARTITION_SPEC — an identity transform on event_date (field ID 6). The warehouse data directories will use event_date=YYYY-MM-DD/ naming — Iceberg’s internal layout. This is separate from the source Parquet layout at data/parquet/date=*/; the ingestion step renames the source date column to Iceberg’s event_date.
• create_or_load_events() — idempotent. Runs the namespace-and-table creation on first call; on subsequent calls, loads and returns the existing table.

Step 2 — Create the table

cd data-engineering/sample-pipeline
python -c "from iceberg.tables import create_or_load_events; t = create_or_load_events(); print(t)"
ls -R warehouse/

Expected:

• warehouse/catalog.db — the SQLite catalog file.
• warehouse/default/events/metadata/v1.metadata.json — initial table metadata.
• No data files yet — the table exists in the catalog but has no rows.

Step 3 — Append the L02 partitions (first snapshot)

python -m iceberg.write_from_parquet

Expected output: a line like:

appended 142389 rows; current_snapshot_id = 4812765309123456789

Inspect the warehouse to see what appeared:

ls warehouse/default/events/data/
ls warehouse/default/events/metadata/

You should now see event_date=…/ subdirectories under data/ and a new v2.metadata.json plus at least one snap-….avro manifest file under metadata/.

Step 4 — Query via DuckDB

python -m iceberg.query_duckdb

Expected: iceberg events row count = 142389 (or whatever your row count was) — same data, two readers: Parquet-direct in L09; Iceberg-through-DuckDB now.

You can also open an interactive Python session and explore:

from iceberg.query_duckdb import open_with_iceberg

with open_with_iceberg() as con:
    rows = con.execute(
        "SELECT type, count(*) AS n FROM events GROUP BY type ORDER BY n DESC LIMIT 5"
    ).fetchall()
    for event_type, count in rows:
        print(f"{event_type}: {count}")

Step 5 — Take a second snapshot

Append the same partitions again. In a real pipeline this would be a second day’s batch; here we are duplicating the data to observe the snapshot mechanism:

python -m iceberg.write_from_parquet

Now list the snapshots:

from iceberg.tables import create_or_load_events

t = create_or_load_events()
for s in t.snapshots():
    print(s.snapshot_id, s.timestamp_ms)

Expected: two snapshot IDs ordered by timestamp. The row count in the current snapshot is now double the first append.

Step 6 — Time travel between snapshots

Read the table as of the first snapshot only:

from iceberg.tables import create_or_load_events

t = create_or_load_events()
snapshots = t.snapshots()
first_snap = snapshots[0]

scan = t.scan(snapshot_id=first_snap.snapshot_id)
arrow_first = scan.to_arrow()
print(f"rows at first snapshot: {arrow_first.num_rows}")

arrow_current = t.scan().to_arrow()
print(f"rows at current snapshot: {arrow_current.num_rows}")

Expected: rows at first snapshot is half of rows at current snapshot. On a real pipeline this lets you replay “what did the table look like before yesterday’s broken DAG?” with no backup tape — just a snapshot ID.

Step 7 — Schema evolution

Add a column and observe how old snapshots handle it:

from pyiceberg.types import LongType
from iceberg.tables import create_or_load_events

t = create_or_load_events()
first_snap = t.snapshots()[0]

# Add a new column (no data rewrite needed):
with t.update_schema() as upd:
    upd.add_column("commit_count", LongType())

# Old snapshot returns NULL for the new column:
arrow_old = t.scan(snapshot_id=first_snap.snapshot_id).to_arrow()
print("Column names:", arrow_old.schema.names)
print("commit_count sample:", arrow_old.column("commit_count")[:5].to_pylist())

Expected: commit_count appears in the column list; its values are all None for the older snapshot. New rows written after the schema change would populate it.

Step 8 — Expire snapshots

python -m iceberg.maintenance expire --retention-days 0

--retention-days 0 forces expiration of everything older than right now — safe for this sample lab where you can re-run the append steps, but never use this on a production table. In production, set retention to at least your longest backfill window.

List snapshots again:

from iceberg.tables import create_or_load_events
t = create_or_load_events()
print([s.snapshot_id for s in t.snapshots()])

Expected: only the most-recent snapshot remains. Data files belonging to expired snapshots are now subject to garbage collection (pyiceberg will delete them on the next maintenance run or when the catalog is compacted).

What you should leave with

Troubleshooting