Project: Kiosko's first Iceberg table

Description

This project closes the module. You have PyIceberg installed (lesson 4), you know how to create a catalog, a namespace, and a table with an explicit schema (lessons 4 and 5), you know how to load real data with table.append() (lesson 6), and you know how to verify the result is correct, not just that "it didn't throw any error" (lesson 7). One step is left: bringing the five pieces together in a single script, run end to end, that reproduces — with your own hands, on your own machine — the exact moment Kiosko stopped having "a Parquet file" and started having "a real Iceberg table."

Connection to the module. This project doesn't introduce any new concept — it's the final integration of the seven previous lessons. It literally revisits the question that opened this module in lesson 1: what does a loose Parquet file lack to behave like a table? This project doesn't close that question completely yet — that takes all eight modules of this guide — but it delivers the first executable answer: a real table, with a real snapshot, verified against the same number six previous guides already confirmed.

An analogy: the complete album, from the blank cover to the first archived collection

Lessons 4 through 7 of this module built, one piece at a time, Kiosko's complete album: the hired librarian (lesson 4), the printed cover and index (lesson 5), the first photo collection pasted in (lesson 6), and the confirmation that those photos really are the right ones (lesson 7). This project is the moment to repeat the whole process, end to end, in a single continuous gesture — like assembling the entire album in front of a client, with no pauses between steps, to prove the whole process works as one coherent piece, not as five steps that only work in isolation.

The material: everything this module built, in one place

You need, in a new working directory:

kiosko_iceberg/
├── raw_orders.py                    (lesson 6: the fixed week of 40 orders)
└── kiosko_first_iceberg_table.py    (this project: brings the 5 pieces together)

With PyIceberg installed in your environment (pip install "pyiceberg[sql-sqlite,pyarrow]", lesson 4).

The reference solution, verified

# kiosko_first_iceberg_table.py -- module 1 closing project
# from a loose Parquet file to Kiosko's first real Iceberg table
import os
from datetime import datetime

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

from raw_orders import RAW_ORDERS

DIM_STORE = [
    {"store_id": "S01", "store_name": "Kiosko Centro", "city": "Bogota"},
    {"store_id": "S02", "store_name": "Kiosko Norte", "city": "Lima"},
    {"store_id": "S03", "store_name": "Kiosko Sur", "city": "Santiago"},
]
DIM_PRODUCT = [
    {"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},
]
store_names = {s["store_id"]: s["store_name"] for s in DIM_STORE}


def build_fact_orders_parquet(path: str) -> None:
    store_ids = {s["store_id"] for s in DIM_STORE}
    product_ids = {p["product_id"] for p in DIM_PRODUCT}

    cols = {"order_id": [], "store_id": [], "product_id": [], "quantity": [],
            "unit_price": [], "revenue": [], "order_ts": []}

    for order_id, store_id, product_id, quantity, unit_price, ts in RAW_ORDERS:
        if store_id not in store_ids:
            raise ValueError(f"unknown store_id: {store_id}")
        if product_id not in product_ids:
            raise ValueError(f"unknown product_id: {product_id}")
        cols["order_id"].append(order_id)
        cols["store_id"].append(store_id)
        cols["product_id"].append(product_id)
        cols["quantity"].append(quantity)
        cols["unit_price"].append(unit_price)
        cols["revenue"].append(round(quantity * unit_price, 10))
        cols["order_ts"].append(datetime.fromisoformat(ts))

    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),
    ])
    pa_table = pa.Table.from_arrays(
        [
            pa.array(cols["order_id"], type=pa.string()),
            pa.array(cols["store_id"], type=pa.string()),
            pa.array(cols["product_id"], type=pa.string()),
            pa.array(cols["quantity"], type=pa.int32()),
            pa.array(cols["unit_price"], type=pa.float64()),
            pa.array(cols["revenue"], type=pa.float64()),
            pa.array(cols["order_ts"], type=pa.timestamp("us")),
        ],
        schema=schema,
    )
    pq.write_table(pa_table, path)


def main() -> None:
    print("=== Kiosko: from fact_orders.parquet to the first Iceberg table ===\n")

    build_fact_orders_parquet("fact_orders.parquet")
    print("Step 1/5 -- fact_orders.parquet reconstructed with pyarrow")

    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}",
    )
    print(f"Step 2/5 -- catalog '{catalog.name}' loaded ({type(catalog).__name__})")

    catalog.create_namespace("kiosko")
    print(f"Step 3/5 -- namespace created: {catalog.list_namespaces()}")

    fact_orders_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),
    )
    table = catalog.create_table("kiosko.fact_orders", schema=fact_orders_schema)
    print(f"Step 4/5 -- table created: {table.name()} (snapshot={table.current_snapshot()})")

    pa_table = pq.read_table("fact_orders.parquet")
    table.append(pa_table)
    snap_id = table.current_snapshot().snapshot_id
    print(f"Step 5/5 -- table.append() completed, snapshot_id captured: {snap_id}\n")

    scanned = table.scan().to_arrow()
    total_rows = scanned.num_rows
    total_revenue = round(sum(scanned.column("revenue").to_pylist()), 2)

    print("=== Final verification ===\n")
    print(f"len(table.scan().to_arrow()) = {total_rows}")

    by_store = scanned.group_by("store_id").aggregate([("revenue", "sum")])
    print("\nRevenue by store:")
    for row in sorted(by_store.to_pylist(), key=lambda r: r["store_id"]):
        sid = row["store_id"]
        print(f"  {sid} {store_names[sid]:<14}: revenue={round(row['revenue_sum'], 2)}")
    print(f"\nTotal week revenue: {total_revenue}")

    keys = pc.binary_join_element_wise(scanned.column("order_id"), scanned.column("product_id"), "-")
    distinct_keys = pc.count_distinct(keys).as_py()
    print(f"\nGrain: total_rows={total_rows}, distinct_order_product_lines={distinct_keys}")

    assert total_rows == 40, f"expected 40 rows, got {total_rows}"
    assert total_revenue == 106.15, f"expected 106.15, got {total_revenue}"
    assert distinct_keys == 40, f"expected 40 distinct combinations, got {distinct_keys}"
    print("\nAll checks passed: 40 rows, 106.15 revenue, 0 duplicates.")


if __name__ == "__main__":
    main()

(raw_orders.py is exactly the same file with the forty fixed orders from lesson 6 — not repeated here for space.)

What to expect (verified by running the actual python3 kiosko_first_iceberg_table.py, end to end, in a new directory; the snapshot_id is the one your run assigns at commit time — different every time, shown here as a placeholder):

=== Kiosko: from fact_orders.parquet to the first Iceberg table ===

Step 1/5 -- fact_orders.parquet reconstructed with pyarrow
Step 2/5 -- catalog 'kiosko' loaded (SqlCatalog)
Step 3/5 -- namespace created: [('kiosko',)]
Step 4/5 -- table created: ('kiosko', 'fact_orders') (snapshot=None)
Step 5/5 -- table.append() completed, snapshot_id captured: <snapshot-id assigned in your run, different each time>

=== Final verification ===

len(table.scan().to_arrow()) = 40

Revenue by store:
  S01 Kiosko Centro : revenue=38.3
  S02 Kiosko Norte  : revenue=38.8
  S03 Kiosko Sur    : revenue=29.05

Total week revenue: 106.15

Grain: total_rows=40, distinct_order_product_lines=40

All checks passed: 40 rows, 106.15 revenue, 0 duplicates.

Notice the script's final assert statements: they're not decorative — if the total revenue weren't exactly 106.15, or if the row count weren't 40, or if the grain had any duplicate, the script would end with an AssertionError instead of the success message. This is the same "verify, don't trust" discipline lesson 7 explained in its Going deeper section, now turned into an automatic check that runs every time you execute this project.

Diagram: where you came from, where you landed

flowchart LR
    A["Lessons 1-3:\nthe problem and the vocabulary\n(file vs table)"] --> B["Lesson 4:\nPyIceberg installed,\ncatalog loaded"]
    B --> C["Lesson 5:\nnamespace + table\nwith schema, empty"]
    C --> D["Lesson 6:\ntable.append()\nfirst real snapshot"]
    D --> E["Lesson 7:\n106.15 verified,\ngrain with no duplicates"]
    E --> F["This project:\nthe 5 pieces, one script,\nautomatic assert"]
    F --> G["Module 2:\ntable anatomy\n(what's really on disk)"]

On disk, after this project

kiosko_warehouse/kiosko/fact_orders/
├── data/
│   └── 00000-0-<uuid>.parquet
└── metadata/
    ├── 00000-<uuid>.metadata.json   (empty table, step 4)
    ├── 00001-<uuid>.metadata.json   (first snapshot, step 5)
    ├── <uuid>-m0.avro               (manifest file)
    └── snap-<snapshot_id>-0-<uuid>.avro   (manifest list)

Exactly the same structure you already saw in lesson 6 — this project doesn't add any new file to the warehouse, it just reproduces the same flow with the five pieces together in a single script, instead of three separate scripts.

Closing lesson 1's promise, point by point

What lesson 1 promised	Evidence this module delivered it
Name the four times Parquet alone wasn't enough	Lesson 2: code quoted literally from foundations, data-modeling, dbt, and spark
Precisely define file format vs. table format	Lesson 3: the distinction and the six capabilities Iceberg adds
Install PyIceberg for real, local, no JVM	pyiceberg==0.11.1 installed and verified (lesson 4)
Create a catalog, a namespace, and a table with a real schema	SqlCatalog + kiosko + kiosko.fact_orders with 7 typed columns (lessons 4-5)
Load Kiosko's fact_orders.parquet into Iceberg	table.append() executed, first real snapshot (lesson 6)
Verify revenue is still 106.15	S01=38.3/S02=38.8/S03=29.05, total 106.15, no duplicates (lesson 7, and this project)

Not a single row of this table solves any of lesson 2's four problems in full yet — that only starts in module 2 (anatomy) and gets solved, one by one, in modules 3 through 6. What this module delivers is the foundation: a real Iceberg table, loaded, verified, on top of which the rest of this guide builds each specific guarantee.

Common mistakes

Running the project against a table that already exists from an earlier lesson, and running into TableAlreadyExistsError. What happens: someone runs this project in the same directory where they already completed lessons 4 through 7, and catalog.create_table("kiosko.fact_orders", ...) fails because the table is already registered in kiosko_catalog.db. Why it happens: this project deliberately repeats the same creation steps you already ran before, so it's self-contained and reproducible from scratch. How to spot it: if you see TableAlreadyExistsError (or an equivalent message) when running kiosko_first_iceberg_table.py, you already have a catalog with that table registered in the same directory. How to fix it: run this project in a new working directory, separate from where you did lessons 4 through 7 — as this lesson's "The material" section suggests — or delete kiosko_catalog.db and kiosko_warehouse/ from the earlier directory before repeating the project there.

Running the script without raw_orders.py in the same directory. What happens: someone copies only kiosko_first_iceberg_table.py, without raw_orders.py, and the script immediately fails with ModuleNotFoundError: No module named 'raw_orders'. Why it happens: the import at the top of the script expects to find that file in the same working directory (or on the PYTHONPATH). How to spot it: the error shows up before even the first line of main() prints — if your output doesn't show the === Kiosko: ... === header at all, check that raw_orders.py exists alongside the script. How to fix it: copy raw_orders.py from lesson 6 into the same directory as kiosko_first_iceberg_table.py, exactly as this lesson's "The material" section shows.

Interpreting the final assert statements as optional, and removing them "to make it run faster." What happens: someone, adapting this script for their own use, deletes the three assert lines at the end, thinking they're just a formality. Why it happens: the assert statements don't change what's shown on screen if everything goes well — the success message shows up either way — so they can feel redundant. How to spot it: without the assert statements, a future bug (for example, if someone modifies RAW_ORDERS without noticing) would produce silently incorrect output, with no warning at all. How to fix it: keep the assert statements — they are, precisely, the difference between "the script ran with no errors" and "the script confirmed the result is correct," the same distinction that motivated all of lesson 7.

Exercises

Exercise 1 — Run the whole project yourself, from scratch. In a new directory, with only raw_orders.py and kiosko_first_iceberg_table.py, run python3 kiosko_first_iceberg_table.py. Confirm you see the five steps complete and the final "All checks passed" message.

See solution

If raw_orders.py is in the same directory and PyIceberg is installed (lesson 4), the output should reproduce exactly this lesson's structure: five numbered steps, followed by the final verification with a 106.15 total and 40 == 40 in the grain check. Your snapshot_id in Step 5/5 is going to be a different integer than anyone else's run — that's expected, not an error.

Exercise 2 — Break the assert on purpose, and watch it fail. Temporarily change the line assert total_revenue == 106.15, ... to assert total_revenue == 999.99, ..., run the script again, and observe what happens. Then revert the change.

See solution

The script should fail with AssertionError: expected 999.99, got 106.15 (the message includes the actual value, 106.15, so it's clear what the assert expected versus what it actually found) — execution stops there, never reaching the final success message. This exercise demonstrates, with direct evidence, that this script's assert statements serve a real purpose: if the business number doesn't match what's expected, the script warns you loudly and immediately, instead of silently finishing with an incorrect result nobody notices.

Exercise 3 — Explain, in your own words, why this project doesn't "teach anything new" and is still worth doing. In 3-4 sentences, justify why integrating code you already wrote in lessons 4 through 7, without adding any new concept, is an exercise worth doing — and not just a repetition.

See solution

Writing each piece separately (catalog, namespace, table, load, verification) in different lessons is the right way to learn each concept without cognitive overload — but a real pipeline never runs as five separate scripts someone runs by hand, one by one, in the right order. Integrating the five pieces into a single script with main() and final assert statements is exactly the kind of work a real data engineer does after prototyping each piece separately: turning "I know how to do each step" into "I have an artifact that reproduces the complete result, reliably, every time I run it." It's the same transition you already saw in python-for-data-engineering-guide, when kiosko_pipeline packaged already-tested functions into a single runnable command.

Summary and next step: closing this module

With this project you close module 1. You integrated the five pieces from lessons 4 through 7 — install, catalog, namespace and table, load, verification — into a single script, run end to end, with automatic assert statements confirming 40 rows, 106.15 total revenue, and zero duplicates in the grain. For the first time in this ecosystem, Kiosko has a real Iceberg table: kiosko.fact_orders, with a snapshot, a declared schema, and the same exact data six different engines already confirmed before.

In eight lessons, you took this guide's first step: from a loose Parquet file, with no memory of itself, to a table with a catalog, a schema, and a snapshot. None of lesson 2's four problems is fully solved yet — that's exactly right at this point in the guide. What you have now is the real foundation the rest of this guide builds each specific guarantee on top of.

Where you go next. Module 2 — Anatomy of an Iceberg table — opens the kiosko_warehouse/ directory this module created and explains, file by file, the complete chain: catalog → metadata → manifest list → manifest files → data files. You're going to inspect, both on disk and with PyIceberg's table.inspect.snapshots()/table.inspect.manifests()/table.inspect.files(), exactly what points to what, and why nothing you saw in this module ever gets overwritten.

Resources

• PyIceberg — official documentation (quickstart), the full install, catalog, namespace, table, and load flow this project integrates. py.iceberg.apache.org. In English.
• PyIceberg — API reference, load_catalog(), create_namespace(), create_table(), table.append(), table.scan(). py.iceberg.apache.org/api. In English.
• Apache Iceberg — official documentation, reference version 1.11.0, table concepts as top-level documented topics. iceberg.apache.org/docs/latest. In English.
• This guide's DESIGN doc — the full map of the eight modules, including module 2 which follows. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.