Project: Kiosko's maintained table

Description

This project closes module 7. You learned about the four production catalogs, without implementing any of them (lesson 2). You measured, with real evidence, how five nights of a redundant pipeline multiplied kiosko.dim_product's snapshot count by more than four (lesson 3). You saw the distinct small-files problem, on kiosko.fact_orders_daily_batches (lesson 4, representative on the compaction side). You really pruned the table with expire_snapshots(), protecting snap_v1 (lesson 5). You identified the orphan files that pruning left behind (lesson 6, representative on the deletion side). And you contrasted all of this, once, with Delta Lake (lesson 7). Only one step is left: bringing this module's executable pieces together in a single script, with automated asserts confirming every number.

Connection to the module. This project doesn't introduce any new concept — it's the final integration of the seven previous lessons. Unlike other closing projects in this guide, this one is explicit about a double boundary: it rebuilds and verifies end to end everything that did really run in this module (expire_snapshots, the snapshot accumulation, the orphan identification), and it records, without faking their execution, the two operations that stayed representative (compaction, remove_orphan_files).

An analogy: the maintenance report, with what got done and what's pending in separate columns

An auto shop handing back a vehicle after a checkup doesn't mix, in one vague paragraph, what it did with what it recommends for the next visit. It hands over a work order with two clear columns: "done today" — with the mechanic's signature and the parts replaced — and "pending, recommended" — with the exact diagnosis and what it would take to fix it. This project is that work order: one column with expire_snapshots() really executed, eleven snapshots pruned, verified with assert; another column with the compaction and remove_orphan_files this environment can't run, documented with the same precision, without pretending they've already been done.

The material: a new working directory

kiosko_maintained_table_project/
├── raw_orders_by_day.py                     (the same 40 orders as always, grouped by day)
└── kiosko_maintained_table_project.py       (this project: brings the executable pieces together)

With PyIceberg installed in your environment (pip install "pyiceberg[sql-sqlite,pyarrow]", module 1, lesson 4). This project is self-contained: it creates kiosko.dim_product and kiosko.fact_orders_daily_batches from scratch, so it runs in a new directory, separate from this module's lessons 3 through 6.

The reference solution, verified

# kiosko_maintained_table_project.py -- module 7's closing project
import os
from datetime import datetime

import pyarrow as pa
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import DoubleType, IntegerType, NestedField, StringType, TimestampType

from raw_orders_by_day import RAW_ORDERS_BY_DAY


def main() -> None:
    print("=== Kiosko: kiosko.dim_product, maintained end to end ===\n")

    warehouse_path = os.path.abspath("kiosko_warehouse")
    catalog_db_path = os.path.abspath("kiosko_catalog.db")
    os.makedirs(warehouse_path, exist_ok=True)
    catalog = load_catalog(
        "kiosko", type="sql",
        uri=f"sqlite:///{catalog_db_path}", warehouse=f"file://{warehouse_path}",
    )
    catalog.create_namespace("kiosko")

    dim_product_schema = Schema(
        NestedField(field_id=1, name="product_id", field_type=StringType(), required=True),
        NestedField(field_id=2, name="product_name", field_type=StringType(), required=True),
        NestedField(field_id=3, name="category", field_type=StringType(), required=True),
        NestedField(field_id=4, name="unit_cost", field_type=DoubleType(), required=True),
    )
    table = catalog.create_table("kiosko.dim_product", schema=dim_product_schema)

    pa_schema = pa.schema([
        pa.field("product_id", pa.string(), nullable=False),
        pa.field("product_name", pa.string(), nullable=False),
        pa.field("category", pa.string(), nullable=False),
        pa.field("unit_cost", pa.float64(), nullable=False),
    ])
    DIM_PRODUCT_V1 = [
        {"product_id": "P001", "product_name": "Bottled Water 600ml", "category": "beverages", "unit_cost": 0.40},
        {"product_id": "P002", "product_name": "Energy Bar", "category": "snacks", "unit_cost": 0.60},
        {"product_id": "P003", "product_name": "Instant Coffee Sachet", "category": "beverages", "unit_cost": 0.35},
        {"product_id": "P004", "product_name": "Phone Charger Cable", "category": "electronics", "unit_cost": 2.10},
    ]
    DIM_PRODUCT_V2 = [
        {"product_id": "P001", "product_name": "Bottled Water 600ml", "category": "beverages", "unit_cost": 0.40},
        {"product_id": "P002", "product_name": "Energy Bar", "category": "health-snacks", "unit_cost": 0.68},
        {"product_id": "P003", "product_name": "Instant Coffee Sachet", "category": "beverages", "unit_cost": 0.35},
        {"product_id": "P004", "product_name": "Phone Charger Cable", "category": "electronics", "unit_cost": 2.10},
    ]

    # Step 1/8 -- rebuild module 3's state: V1 + P002's real change
    table.append(pa.Table.from_pylist(DIM_PRODUCT_V1, schema=pa_schema))
    snap_v1 = table.current_snapshot().snapshot_id
    table.overwrite(pa.Table.from_pylist(DIM_PRODUCT_V2, schema=pa_schema))
    snapshots_after_m3 = len(table.history())
    print(f"Step 1/8 -- module 3's state rebuilt: {snapshots_after_m3} snapshots, snap_v1 captured")

    # Step 2/8 -- 5 redundant nights (this module's lesson 3)
    nights = ["2026-08-16", "2026-08-17", "2026-08-18", "2026-08-19", "2026-08-20"]
    for _ in nights:
        table.overwrite(pa.Table.from_pylist(DIM_PRODUCT_V2, schema=pa_schema))
    snapshots_before_expire = len(table.history())
    all_files_before = table.inspect.all_data_files().num_rows
    live_files_before = table.inspect.files().num_rows
    print(f"Step 2/8 -- 5 redundant nights applied: {snapshots_before_expire} snapshots, "
          f"{all_files_before} tracked files, {live_files_before} live")

    # Step 3/8 -- build the pruning list: everything except snap_v1 and the current one
    history = table.history()
    current_snapshot_id = table.current_snapshot().snapshot_id
    to_expire = [e.snapshot_id for e in history if e.snapshot_id not in (snap_v1, current_snapshot_id)]
    print(f"Step 3/8 -- {len(to_expire)} snapshots selected to expire "
          f"(protected: snap_v1 and the current one)")

    # Step 4/8 -- REAL expire_snapshots()
    table.maintenance.expire_snapshots().by_ids(to_expire).commit()
    table.refresh()
    snapshots_after_expire = len(table.history())
    all_files_after = table.inspect.all_data_files().num_rows
    print(f"Step 4/8 -- expire_snapshots().by_ids() run: {snapshots_after_expire} snapshots, "
          f"{all_files_after} tracked files")

    # Step 5/8 -- verify time travel and the current state stayed intact
    v1_rows = table.scan(snapshot_id=snap_v1).to_arrow().to_pylist()
    p002_v1 = next(r for r in v1_rows if r["product_id"] == "P002")
    current_rows = table.scan().to_arrow().to_pylist()
    p002_current = next(r for r in current_rows if r["product_id"] == "P002")
    print(f"Step 5/8 -- AS OF snap_v1: P002={p002_v1['category']}/{p002_v1['unit_cost']} | "
          f"current: P002={p002_current['category']}/{p002_current['unit_cost']}")

    # Step 6/8 -- identify orphan files left on disk (diagnostic, read-only)
    tracked = {row["file_path"] for row in table.inspect.all_data_files().select(["file_path"]).to_pylist()}
    data_dir = os.path.join(warehouse_path, "kiosko", "dim_product", "data")
    on_disk = {"file://" + os.path.join(data_dir, fn) for fn in os.listdir(data_dir) if fn.endswith(".parquet")}
    orphans = on_disk - tracked
    orphan_bytes = sum(os.path.getsize(f.replace("file://", "")) for f in orphans)
    print(f"Step 6/8 -- {len(on_disk)} physical files on disk, {len(orphans)} orphans "
          f"({orphan_bytes} bytes) -- remove_orphan_files is representative, not run here")

    # Step 7/8 -- kiosko.fact_orders_daily_batches -- the small files problem (lesson 4)
    fact_schema = Schema(
        NestedField(field_id=1, name="order_id", field_type=StringType(), required=True),
        NestedField(field_id=2, name="store_id", field_type=StringType(), required=True),
        NestedField(field_id=3, name="product_id", field_type=StringType(), required=True),
        NestedField(field_id=4, name="quantity", field_type=IntegerType(), required=True),
        NestedField(field_id=5, name="unit_price", field_type=DoubleType(), required=True),
        NestedField(field_id=6, name="revenue", field_type=DoubleType(), required=True),
        NestedField(field_id=7, name="order_ts", field_type=TimestampType(), required=True),
    )
    fact_pa_schema = pa.schema([
        pa.field("order_id", pa.string(), nullable=False),
        pa.field("store_id", pa.string(), nullable=False),
        pa.field("product_id", pa.string(), nullable=False),
        pa.field("quantity", pa.int32(), nullable=False),
        pa.field("unit_price", pa.float64(), nullable=False),
        pa.field("revenue", pa.float64(), nullable=False),
        pa.field("order_ts", pa.timestamp("us"), nullable=False),
    ])
    fact_table = catalog.create_table("kiosko.fact_orders_daily_batches", schema=fact_schema)
    for day, orders in RAW_ORDERS_BY_DAY:
        rows = []
        for order_id, store_id, product_id, quantity, unit_price, ts in orders:
            rows.append({
                "order_id": order_id, "store_id": store_id, "product_id": product_id,
                "quantity": quantity, "unit_price": unit_price,
                "revenue": round(quantity * unit_price, 10),
                "order_ts": datetime.fromisoformat(ts),
            })
        fact_table.append(pa.Table.from_pylist(rows, schema=fact_pa_schema))
    fact_rows = fact_table.scan().to_arrow().to_pylist()
    fact_total_rows = len(fact_rows)
    fact_total_revenue = round(sum(r["revenue"] for r in fact_rows), 2)
    fact_live_files = fact_table.inspect.files().num_rows
    print(f"Step 7/8 -- kiosko.fact_orders_daily_batches: {fact_total_rows} rows, "
          f"revenue={fact_total_revenue}, {fact_live_files} live files "
          f"(compaction is representative, not run here)")

    print("\n=== Final verification ===\n")

    assert snapshots_after_m3 == 3, "module 3 should leave 3 snapshots (append, delete, append)"
    assert snapshots_before_expire == 13, "5 redundant nights (2 snapshots each) + 3 = 13"
    assert len(to_expire) == 11
    assert snapshots_after_expire == 2, "only snap_v1 and the current one should survive"
    assert all_files_before == 7
    assert all_files_after == 2
    assert p002_v1["category"] == "snacks" and p002_v1["unit_cost"] == 0.60
    assert p002_current["category"] == "health-snacks" and p002_current["unit_cost"] == 0.68
    assert len(orphans) == 5, "the 5 files expire_snapshots() stopped tracking are still on disk"
    assert fact_total_rows == 40
    assert fact_total_revenue == 106.15
    assert fact_live_files == 7, "7 daily append()s, none redundant, not compacted"

    print("All verifications passed:")
    print(f"  - kiosko.dim_product: {snapshots_after_m3} -> {snapshots_before_expire} snapshots "
          f"(5 redundant nights) -> {snapshots_after_expire} (post expire_snapshots, REAL)")
    print("  - snap_v1 protected: P002 is still snacks/0.60 via time travel")
    print("  - current intact: P002 is still health-snacks/0.68")
    print(f"  - {len(orphans)} orphan files identified ({orphan_bytes} bytes), "
          "remove_orphan_files stays representative")
    print(f"  - kiosko.fact_orders_daily_batches: {fact_total_rows} rows, revenue={fact_total_revenue}, "
          f"{fact_live_files} live files, compaction stays representative")


if __name__ == "__main__":
    main()

(raw_orders_by_day.py groups the same forty orders from module 1, lesson 6, by the real day each one happened — not repeated here for space; it's exactly the dictionary you already saw in full in this module's lesson 4.)

What to expect (verified by running the real python3 kiosko_maintained_table_project.py, end to end, in a new directory; no snapshot_id gets printed as a literal):

=== Kiosko: kiosko.dim_product, maintained end to end ===

Step 1/8 -- module 3's state rebuilt: 3 snapshots, snap_v1 captured
Step 2/8 -- 5 redundant nights applied: 13 snapshots, 7 tracked files, 1 live
Step 3/8 -- 11 snapshots selected to expire (protected: snap_v1 and the current one)
Step 4/8 -- expire_snapshots().by_ids() run: 2 snapshots, 2 tracked files
Step 5/8 -- AS OF snap_v1: P002=snacks/0.6 | current: P002=health-snacks/0.68
Step 6/8 -- 7 physical files on disk, 5 orphans (9185 bytes) -- remove_orphan_files is representative, not run here
Step 7/8 -- kiosko.fact_orders_daily_batches: 40 rows, revenue=106.15, 7 live files (compaction is representative, not run here)

=== Final verification ===

All verifications passed:
  - kiosko.dim_product: 3 -> 13 snapshots (5 redundant nights) -> 2 (post expire_snapshots, REAL)
  - snap_v1 protected: P002 is still snacks/0.60 via time travel
  - current intact: P002 is still health-snacks/0.68
  - 5 orphan files identified (9185 bytes), remove_orphan_files stays representative
  - kiosko.fact_orders_daily_batches: 40 rows, revenue=106.15, 7 live files, compaction stays representative

Eleven asserts, none decorative: they confirm the exact snapshot count at each stage (3 → 13 → 2), that the pruning list was exactly as expected (11), that time travel and the current state survived pruning intact, that the identified orphan files match what lesson 6 predicted (5, 9185 bytes), and that lesson 4's small-files problem is still reproducible (7 live files, 40 rows, 106.15 in revenue).

Diagram: where you came from, where you landed

flowchart LR
    A["Modules 1-6:\nreal kiosko.dim_product,\nsnap_v1 protected"] --> B["Lesson 2:\n4 production catalogs\nnamed, not implemented"]
    B --> C["Lesson 3:\n5 redundant nights,\n3 -> 13 snapshots"]
    C --> D["Lesson 4:\nsmall files,\nfact_orders_daily_batches"]
    D --> E["Lesson 5:\nREAL expire_snapshots(),\n13 -> 2 snapshots"]
    E --> F["Lesson 6:\n5 orphans identified,\nremove_orphan_files representative"]
    F --> G["Lesson 7:\nDelta Lake by contrast,\n2026 convergence"]
    G --> H["This project:\neverything integrated,\n11 automated assert"]
    H --> I["Module 8:\nKiosko's lakehouse\ncapstone"]

Closing the module's promise, point by point

What lesson 1 promised	Evidence this module delivered it
Naming the production catalogs, without implementing them	Lesson 2: REST, Glue, Unity Catalog, Polaris, each with its guarantee quoted against official documentation
Measuring why snapshots accumulate cost	Lesson 3 and this project: 3 → 13 snapshots, 7 tracked files, 1 live, verified with assert
Compacting small files	Lesson 4: kiosko.fact_orders_daily_batches, 7 real live files, compaction documented as representative (Spark, rewrite_data_files)
Safely expiring old snapshots	Lesson 5 and this project: expire_snapshots().by_ids() really executed, snap_v1 protected, time travel verified intact
Removing orphan files without losing the time travel you do need	Lesson 6 and this project: 5 orphans identified with real code (9185 bytes), remove_orphan_files documented as representative
Delta Lake named once, by contrast	Lesson 7: metadata mechanism, time travel syntax, maintenance, and the 2026 convergence — with no Delta table ever built

This project left no lesson 1 promise without evidence — including the two this environment couldn't really execute, documented with the same honesty the rest of this guide applied to Spark's MERGE INTO in module 6.

Common mistakes

Running this project on a catalog that already has kiosko.dim_product or kiosko.fact_orders_daily_batches from an earlier lesson in this module. What happens: someone runs this project in the same directory where they already completed lessons 3 through 6, and catalog.create_table(...) fails because the tables are already registered. Why it happens: this project deliberately repeats the full rebuild from scratch, so it's self-contained and reproducible without depending on the exact state previous lessons left behind. How to spot it: if you see TableAlreadyExistsError when running kiosko_maintained_table_project.py, you already have a catalog with those tables registered in the same directory. How to fix it: run this project in a new working directory, separate from where you did lessons 3 through 6 — exactly as this lesson's "The material" suggests.

Assuming all_files_after == 2 in Step 4 means disk space is already freed. What happens: someone, seeing Step 4's assert pass with all_files_after == 2, concludes the project already completed all possible cleanup. Why it happens: it's easy, again, not to distinguish between "files the metadata tracks" (what expire_snapshots() does change) and "physical files on disk" (what only remove_orphan_files would change). How to spot it: check Step 6's assert len(orphans) == 5 — that same script, running immediately after pruning, confirms five physical files with no reference at all still exist. How to fix it: read both asserts together, not separately — together they tell the whole story: the metadata is already pruned (2 tracked files), but the disk still has pending work (5 orphans), exactly the distinction this module's lessons 5 and 6 developed in depth.

Exercises

Exercise 1 — Run the full project yourself, from scratch. In a new directory, with raw_orders_by_day.py in the same place, run python3 kiosko_maintained_table_project.py. Confirm you see the seven steps complete and the final message with the five verifications.

See solution

If raw_orders_by_day.py is in the same directory and PyIceberg is installed, the output should exactly reproduce this lesson's structure: seven numbered steps, followed by the final verification with the five success messages. Your run's snapshot_ids are going to be different from any earlier example in this guide — that's exactly what's expected.

Exercise 2 — Break an assert on purpose, and watch it fail. Temporarily change the number of redundant nights from five to three (nights = ["2026-08-16", "2026-08-17", "2026-08-18"]), run the script again, and observe which assert fails first. Then revert the change.

See solution

The first assert to fail should be assert snapshots_before_expire == 13 — with three nights instead of five, each producing two snapshots, the total after Step 2 would be 3 + 3*2 = 9, not 13. If you fixed that number to 9, the next one to fail would be assert len(to_expire) == 11 (now there'd be 9 - 2 = 7 candidates), and, cascading down, assert all_files_before == 7 would also change. This exercise confirms this project's numbers are precisely chained to the exact number of redundant writes — any change in Step 2 propagates, predictably, to every later verification.

Exercise 3 — Explain, in your own words, why this project includes both kiosko.dim_product (the old-snapshots problem) and kiosko.fact_orders_daily_batches (the small-files problem) in the same script, instead of limiting itself to just one. In 3-4 sentences, justify this decision with what you learned in lessons 3 and 4.

See solution

Limiting itself to a single problem would leave the project incomplete against what this module's lesson 1 promised: two distinct cost axes, not one. Lesson 3 and lesson 4 of this module were explicit that "old snapshots nobody needs anymore" and "small files inside the current snapshot" are orthogonal problems — a table can have one, the other, both, or neither — and confusing them is exactly one of the common mistakes lesson 4 documented. Including both in the same closing project, with separate asserts for each, confirms that whoever completes this module can recognize and diagnose the two problems independently, instead of assuming solving one automatically solves the other.

Summary and next step: this module's close

With this project you close module 7. You integrated, in a single script with eleven automated asserts, everything this module really executed: the real snapshot accumulation (3 → 13), the real pruning with snap_v1 protected (13 → 2), the real orphan file identification (5, 9185 bytes), and the real small-files problem on a fact table loaded day by day (7 live files). And you explicitly recorded, without faking their execution, the two operations this environment can't run in pure Python: compaction and remove_orphan_files, both documented with their exact Spark syntax.

Kiosko now has a table that doesn't just know how to recover its history with time travel (module 3) — it also knows how to prune the part of that history that no longer adds anything, without risking the part that does.

Where you go next. Module 8 — this guide's capstone — assembles Kiosko's complete lakehouse: fact_orders, dim_store (with country), dim_product (historized via time travel, with no columns), dim_date, and fact_orders_at_scale (partitioned and evolved), all together, with the same total revenue (106.15) and the same correct P002 margin (10.8) data-modeling-for-analytics-guide and dbt-analytics-engineering-guide already confirmed. And it closes with the map toward this ecosystem's sibling guides — including this one, with its production catalogs still unimplemented and its compaction still representative, the two exact boundaries aws-core-services-guide picks up.

Resources

• PyIceberg — official documentation (quickstart), the complete catalog, table, append(), and overwrite() flow this project integrates. py.iceberg.apache.org. In English.
• PyIceberg — API reference, table.maintenance, table.inspect.all_data_files(), table.inspect.files(), the foundation of this project's Steps 3 through 6. py.iceberg.apache.org/api. In English.
• Apache Iceberg — official documentation, "Maintenance," the source for the three operations this whole module organized. iceberg.apache.org/docs/latest/maintenance. In English.
• This same guide, module 3, lesson 8 — source of the closing-project pattern with chained asserts this project reuses. ../module-03-snapshots-and-time-travel/en/08-project-kioskos-time-traveled-dim-product.md. In English.
• This guide's DESIGN doc — the full map of the eight modules, including the module 8 that follows. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.