Project: Kiosko's first lakehouse

Description

This project closes module 8 — and with it, closes the entire guide. You have Kiosko's lakehouse's five tables built, lesson by lesson, in the same catalog: fact_orders and dim_store with country and dim_date (lesson 3), dim_product with no history columns with P002's correct margin recovered via time travel (lesson 4), fact_orders_at_scale partitioned and evolved (lesson 5), and confirmation that table.upsert() produces the same result as table.overwrite() (lesson 6). You know, with quoted evidence, which seven boundaries this lakehouse leaves pending and which sibling guide solves each one (lesson 7). Only one step is left: rebuilding the five tables, from scratch, in a single script, with automated asserts confirming every number — the same closing pattern each of this guide's seven previous modules already used, now applied to the complete lakehouse.

Connection to the module. This project doesn't introduce any new concept — it's the final integration of this module's seven previous lessons, and of the whole guide's seven previous modules. It literally revisits the promise that opened this guide in its module 1: turning a loose Parquet file into a real table, with a catalog, schema, and snapshot — now multiplied by five tables, coexisting in the same lakehouse, with the same revenue (106.15) and the same correct P002 margin (10.8) data-modeling-for-analytics-guide and dbt-analytics-engineering-guide already confirmed.

An analogy: the complete model, presented all at once

This module's lessons 3 through 6 built, one piece at a time, Kiosko's lakehouse's five tables: the foundations and structure (lesson 3), the central piece with its own recoverable history (lesson 4), the at-scale wing (lesson 5), and confirmation that two different construction paths reach the same result (lesson 6). This project is the moment to repeat the entire process, end to end, in a single continuous gesture — the same final integration that already closed this guide's modules 1, 3, 4, 5, 6, and 7, now applied to the complete lakehouse, not to a single table.

The material: everything this module built, in one place

You need, in a new working directory:

kiosko_first_lakehouse/
├── raw_orders.py                    (module 1, lesson 6: the fixed 40-order week)
├── kiosko_scale.py                  (module 5: the deterministic at-scale generator)
└── kiosko_first_lakehouse.py        (this project: brings the 5 tables together)

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

Scale note. This project loads fact_orders_at_scale's full 10,000,000 rows — the same 250,000 franchises as always; on a modern laptop, the complete script takes between thirty seconds and a minute. The slow part is, by far, generating and loading that table — the rest runs almost instantly.

The reference solution, verified

# kiosko_first_lakehouse.py -- this guide's closing project (module 8)
# Kiosko's complete lakehouse on Apache Iceberg, assembled end to end
import os
from collections import defaultdict
from datetime import date, datetime, timedelta

import pyarrow as pa
import pyarrow.compute as pc
from pyiceberg.catalog import load_catalog
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.transforms import DayTransform, IdentityTransform
from pyiceberg.types import (
    BooleanType, DateType, DoubleType, IntegerType, NestedField, StringType, TimestampType,
)

from kiosko_scale import generate_orders_at_scale
from raw_orders import RAW_ORDERS

NUM_FRANCHISES = 250_000
DAY_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]

DIM_STORE_ROWS = [
    {"store_id": "S01", "store_name": "Kiosko Centro", "city": "Bogota", "country": "Colombia"},
    {"store_id": "S02", "store_name": "Kiosko Norte", "city": "Lima", "country": "Peru"},
    {"store_id": "S03", "store_name": "Kiosko Sur", "city": "Santiago", "country": "Chile"},
]
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},
]

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),
)
DIM_STORE_SCHEMA = Schema(
    NestedField(field_id=1, name="store_id", field_type=StringType(), required=True),
    NestedField(field_id=2, name="store_name", field_type=StringType(), required=True),
    NestedField(field_id=3, name="city", field_type=StringType(), required=True),
    NestedField(field_id=4, name="country", field_type=StringType(), required=True),
)
DIM_DATE_SCHEMA = Schema(
    NestedField(field_id=1, name="date_key", field_type=IntegerType(), required=True),
    NestedField(field_id=2, name="calendar_date", field_type=DateType(), required=True),
    NestedField(field_id=3, name="day_of_week", field_type=StringType(), required=True),
    NestedField(field_id=4, name="month", field_type=IntegerType(), required=True),
    NestedField(field_id=5, name="quarter", field_type=IntegerType(), required=True),
    NestedField(field_id=6, name="year", field_type=IntegerType(), required=True),
    NestedField(field_id=7, name="is_weekend", field_type=BooleanType(), required=True),
)
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),
)
FACT_ORDERS_AT_SCALE_SCHEMA = Schema(
    NestedField(field_id=1, name="order_id", field_type=StringType(), required=True),
    NestedField(field_id=2, name="franchise_id", field_type=IntegerType(), required=True),
    NestedField(field_id=3, name="store_id", field_type=StringType(), required=True),
    NestedField(field_id=4, name="product_id", field_type=StringType(), required=True),
    NestedField(field_id=5, name="quantity", field_type=IntegerType(), required=True),
    NestedField(field_id=6, name="unit_price", field_type=DoubleType(), required=True),
    NestedField(field_id=7, name="order_ts", field_type=TimestampType(), required=True),
)
INITIAL_SPEC = PartitionSpec(
    PartitionField(source_id=3, field_id=1000, transform=IdentityTransform(), name="store_id"),
)

FACT_ORDERS_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),
])
DIM_STORE_PA_SCHEMA = pa.schema([
    pa.field("store_id", pa.string(), nullable=False),
    pa.field("store_name", pa.string(), nullable=False),
    pa.field("city", pa.string(), nullable=False),
    pa.field("country", pa.string(), nullable=False),
])
DIM_DATE_PA_SCHEMA = pa.schema([
    pa.field("date_key", pa.int32(), nullable=False),
    pa.field("calendar_date", pa.date32(), nullable=False),
    pa.field("day_of_week", pa.string(), nullable=False),
    pa.field("month", pa.int32(), nullable=False),
    pa.field("quarter", pa.int32(), nullable=False),
    pa.field("year", pa.int32(), nullable=False),
    pa.field("is_weekend", pa.bool_(), nullable=False),
])
DIM_PRODUCT_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),
])
FACT_AT_SCALE_PA_SCHEMA = pa.schema([
    pa.field("order_id", pa.string(), nullable=False),
    pa.field("franchise_id", pa.int32(), 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("order_ts", pa.timestamp("us"), nullable=False),
])


def fact_orders_pa_table() -> pa.Table:
    rows = [
        {"order_id": oid, "store_id": sid, "product_id": pid, "quantity": qty, "unit_price": price,
         "revenue": round(qty * price, 10), "order_ts": datetime.fromisoformat(ts)}
        for oid, sid, pid, qty, price, ts in RAW_ORDERS
    ]
    return pa.Table.from_pylist(rows, schema=FACT_ORDERS_PA_SCHEMA)


def dim_date_pa_table(start_date: str, end_date: str) -> pa.Table:
    start, end, rows = date.fromisoformat(start_date), date.fromisoformat(end_date), []
    current = start
    while current <= end:
        weekday_index = current.weekday()
        rows.append({
            "date_key": int(current.strftime("%Y%m%d")), "calendar_date": current,
            "day_of_week": DAY_NAMES[weekday_index], "month": current.month,
            "quarter": (current.month - 1) // 3 + 1, "year": current.year,
            "is_weekend": weekday_index >= 5,
        })
        current += timedelta(days=1)
    return pa.Table.from_pylist(rows, schema=DIM_DATE_PA_SCHEMA)


def revenue_of(arrow_table: pa.Table) -> float:
    line_revenue = pc.multiply(pc.cast(arrow_table.column("quantity"), pa.float64()), arrow_table.column("unit_price"))
    return float(pc.sum(line_revenue).as_py())


def rows_to_pa_table(num_franchises: int, franchise_offset: int = 0) -> pa.Table:
    rows = list(generate_orders_at_scale(num_franchises))
    for r in rows:
        r["franchise_id"] += franchise_offset
        if franchise_offset:
            r["order_id"] = r["order_id"].replace(
                f"F{r['franchise_id'] - franchise_offset:06d}", f"F{r['franchise_id']:06d}",
            )
        r["order_ts"] = datetime.fromisoformat(r["order_ts"])
    return pa.Table.from_pylist(rows, schema=FACT_AT_SCALE_PA_SCHEMA)


def margin_by_category(dim_rows: list, fact_rows: list) -> tuple:
    dim_by_id = {r["product_id"]: r for r in dim_rows}
    revenue, margin = defaultdict(float), defaultdict(float)
    for f in fact_rows:
        d = dim_by_id[f["product_id"]]
        revenue[d["category"]] += f["revenue"]
        margin[d["category"]] += f["revenue"] - f["quantity"] * d["unit_cost"]
    return revenue, margin


def main() -> None:
    print("=== Kiosko: the first complete lakehouse, on Apache Iceberg ===\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")
    print(f"Step  1/12 -- catalog '{catalog.name}' and namespace 'kiosko' ready")

    # -- fact_orders --
    fact_orders = catalog.create_table("kiosko.fact_orders", schema=FACT_ORDERS_SCHEMA)
    fact_orders.append(fact_orders_pa_table())
    fact_rows = fact_orders.scan().to_arrow()
    fact_revenue = round(sum(fact_rows.column("revenue").to_pylist()), 2)
    print(f"Step  2/12 -- kiosko.fact_orders: {fact_rows.num_rows} rows, revenue {fact_revenue}")

    # -- dim_store, with country since the first commit --
    dim_store = catalog.create_table("kiosko.dim_store", schema=DIM_STORE_SCHEMA)
    dim_store.append(pa.Table.from_pylist(DIM_STORE_ROWS, schema=DIM_STORE_PA_SCHEMA))
    store_rows = sorted(dim_store.scan().to_arrow().to_pylist(), key=lambda r: r["store_id"])
    print(f"Step  3/12 -- kiosko.dim_store: {len(store_rows)} rows, "
          f"country={[r['country'] for r in store_rows]}")

    # -- dim_date --
    dim_date = catalog.create_table("kiosko.dim_date", schema=DIM_DATE_SCHEMA)
    dim_date.append(dim_date_pa_table("2026-08-01", "2026-08-31"))
    date_rows = dim_date.scan().to_arrow().num_rows
    print(f"Step  4/12 -- kiosko.dim_date: {date_rows} rows (all of August 2026)")

    # -- dim_product: V1, snap_v1, overwrite to V2 --
    dim_product = catalog.create_table("kiosko.dim_product", schema=DIM_PRODUCT_SCHEMA)
    dim_product.append(pa.Table.from_pylist(DIM_PRODUCT_V1, schema=DIM_PRODUCT_PA_SCHEMA))
    snap_v1 = dim_product.current_snapshot().snapshot_id
    print(f"Step  5/12 -- kiosko.dim_product: V1 loaded, snap_v1 captured (P002=snacks/0.60)")

    dim_product.overwrite(pa.Table.from_pylist(DIM_PRODUCT_V2, schema=DIM_PRODUCT_PA_SCHEMA))
    current_p002 = next(r for r in dim_product.scan().to_arrow().to_pylist() if r["product_id"] == "P002")
    print(f"Step  6/12 -- kiosko.dim_product: V2 applied with overwrite() "
          f"(P002={current_p002['category']}/{current_p002['unit_cost']}, effective 2026-08-15)")

    # -- the star joined with time travel: broken margin vs correct --
    fact_rows_list = fact_orders.scan().to_arrow().to_pylist()
    current_rows = dim_product.scan().to_arrow().to_pylist()
    v1_rows = dim_product.scan(snapshot_id=snap_v1).to_arrow().to_pylist()
    revenue_broken, margin_broken = margin_by_category(current_rows, fact_rows_list)
    revenue_correct, margin_correct = margin_by_category(v1_rows, fact_rows_list)
    print(f"Step  7/12 -- joined star (fact_orders + dim_store + dim_date + dim_product): "
          f"broken margin (health-snacks)={round(margin_broken['health-snacks'], 2)}, "
          f"correct margin AS OF snap_v1 (snacks)={round(margin_correct['snacks'], 2)}")

    # -- table.upsert(): the native way, verified against the same result --
    dim_product_check = catalog.create_table("kiosko.dim_product_merge_check", schema=DIM_PRODUCT_SCHEMA)
    dim_product_check.append(pa.Table.from_pylist(DIM_PRODUCT_V1, schema=DIM_PRODUCT_PA_SCHEMA))
    only_p002 = [r for r in DIM_PRODUCT_V2 if r["product_id"] == "P002"]
    upsert_result = dim_product_check.upsert(
        pa.Table.from_pylist(only_p002, schema=DIM_PRODUCT_PA_SCHEMA), join_cols=["product_id"],
    )
    check_rows = sorted(dim_product_check.scan().to_arrow().to_pylist(), key=lambda r: r["product_id"])
    same_result = sorted(current_rows, key=lambda r: r["product_id"]) == check_rows
    print(f"Step  8/12 -- table.upsert() (native way): rows_updated={upsert_result.rows_updated}, "
          f"rows_inserted={upsert_result.rows_inserted}, result identical to overwrite(): {same_result}")

    # -- fact_orders_at_scale: partitioned and evolved --
    fact_at_scale = catalog.create_table(
        "kiosko.fact_orders_at_scale", schema=FACT_ORDERS_AT_SCALE_SCHEMA, partition_spec=INITIAL_SPEC,
    )
    bulk_pa_table = rows_to_pa_table(NUM_FRANCHISES)
    fact_at_scale.append(bulk_pa_table)
    scale_row_count = fact_at_scale.scan().to_arrow().num_rows
    scale_revenue = revenue_of(bulk_pa_table)
    print(f"Step  9/12 -- kiosko.fact_orders_at_scale: {scale_row_count} rows loaded, "
          f"revenue {round(scale_revenue, 2)}")

    s01_scan = fact_at_scale.scan(row_filter="store_id == 'S01'").to_arrow()
    s01_revenue = revenue_of(s01_scan)
    print(f"Step 10/12 -- hidden query store_id == 'S01': {s01_scan.num_rows} rows, "
          f"revenue {round(s01_revenue, 2)}")

    with fact_at_scale.update_spec() as update:
        update.add_field("order_ts", DayTransform(), "order_day")
    new_franchise = rows_to_pa_table(1, franchise_offset=NUM_FRANCHISES)
    fact_at_scale.append(new_franchise)
    partitions = fact_at_scale.inspect.partitions().to_pylist()
    spec_ids_present = sorted({p["spec_id"] for p in partitions})
    scale_row_count_final = fact_at_scale.scan().to_arrow().num_rows
    scale_revenue_final = round(revenue_of(fact_at_scale.scan().to_arrow()), 2)
    print(f"Step 11/12 -- spec evolved (DayTransform over order_ts), new franchise loaded: "
          f"{scale_row_count_final} final rows, revenue {scale_revenue_final}, "
          f"spec_id coexisting: {spec_ids_present}")

    all_tables = sorted(t[1] for t in catalog.list_tables("kiosko"))
    print(f"Step 12/12 -- namespace 'kiosko' complete: {all_tables}\n")

    print("=== Complete lakehouse final verification ===\n")

    assert fact_rows.num_rows == 40
    assert fact_revenue == 106.15
    assert {r["store_id"]: r["country"] for r in store_rows} == {
        "S01": "Colombia", "S02": "Peru", "S03": "Chile",
    }
    assert date_rows == 31
    assert len(dim_product.schema().fields) == 4, "dim_product must not have any history column"
    assert current_p002["category"] == "health-snacks" and current_p002["unit_cost"] == 0.68
    assert round(margin_broken["health-snacks"], 2) == 9.36
    assert round(margin_correct["snacks"], 2) == 10.8
    assert round(revenue_correct["snacks"], 2) == round(revenue_broken["health-snacks"], 2) == 21.6
    assert upsert_result.rows_updated == 1 and upsert_result.rows_inserted == 0
    assert same_result, "overwrite() and upsert() must produce the same business result"
    assert scale_row_count == NUM_FRANCHISES * 40 == 10_000_000
    assert round(scale_revenue, 2) == 26_537_500.00
    assert s01_scan.num_rows == 4_000_000
    assert round(s01_revenue, 2) == 9_575_000.00
    assert spec_ids_present == [0, 1]
    assert scale_row_count_final == 10_000_040
    assert scale_revenue_final == 26_537_606.15
    assert all_tables == [
        "dim_date", "dim_product", "dim_product_merge_check", "dim_store", "fact_orders", "fact_orders_at_scale",
    ]

    print("All verifications passed -- Kiosko's complete lakehouse on Apache Iceberg:")
    print(f"  - kiosko.fact_orders:          40 rows, total revenue {fact_revenue}")
    print(f"  - kiosko.dim_store:            3 rows, country=Colombia/Peru/Chile")
    print(f"  - kiosko.dim_date:             31 rows (August 2026)")
    print(f"  - kiosko.dim_product:          4 columns, zero of history, P002 historized via time travel")
    print(f"      CORRECT margin (AS OF snap_v1, snacks)      = 10.8")
    print(f"      BROKEN margin (current, health-snacks)      = 9.36")
    print(f"      -- identical to data-modeling-for-analytics-guide M8 and dbt-analytics-engineering-guide M8")
    print(f"  - table.upsert() reproduces, row for row, the same result as table.overwrite()")
    print(f"  - kiosko.fact_orders_at_scale: 10,000,040 rows, revenue {scale_revenue_final}, "
          f"2 partition schemes coexisting")


if __name__ == "__main__":
    main()

(raw_orders.py and kiosko_scale.py are exactly the same files from module 1, lesson 6, and from module 5 — not repeated here for space.)

What to expect (verified by running the real python3 kiosko_first_lakehouse.py, end to end, in a new directory; the complete runtime was approximately 32 seconds; no snapshot_id gets printed as a literal — this guide's hard rule, since module 3):

=== Kiosko: the first complete lakehouse, on Apache Iceberg ===

Step  1/12 -- catalog 'kiosko' and namespace 'kiosko' ready
Step  2/12 -- kiosko.fact_orders: 40 rows, revenue 106.15
Step  3/12 -- kiosko.dim_store: 3 rows, country=['Colombia', 'Peru', 'Chile']
Step  4/12 -- kiosko.dim_date: 31 rows (all of August 2026)
Step  5/12 -- kiosko.dim_product: V1 loaded, snap_v1 captured (P002=snacks/0.60)
Step  6/12 -- kiosko.dim_product: V2 applied with overwrite() (P002=health-snacks/0.68, effective 2026-08-15)
Step  7/12 -- joined star (fact_orders + dim_store + dim_date + dim_product): broken margin (health-snacks)=9.36, correct margin AS OF snap_v1 (snacks)=10.8
Step  8/12 -- table.upsert() (native way): rows_updated=1, rows_inserted=0, result identical to overwrite(): True
Step  9/12 -- kiosko.fact_orders_at_scale: 10000000 rows loaded, revenue 26537500.0
Step 10/12 -- hidden query store_id == 'S01': 4000000 rows, revenue 9575000.0
Step 11/12 -- spec evolved (DayTransform over order_ts), new franchise loaded: 10000040 final rows, revenue 26537606.15, spec_id coexisting: [0, 1]
Step 12/12 -- namespace 'kiosko' complete: ['dim_date', 'dim_product', 'dim_product_merge_check', 'dim_store', 'fact_orders', 'fact_orders_at_scale']

=== Complete lakehouse final verification ===

All verifications passed -- Kiosko's complete lakehouse on Apache Iceberg:
  - kiosko.fact_orders:          40 rows, total revenue 106.15
  - kiosko.dim_store:            3 rows, country=Colombia/Peru/Chile
  - kiosko.dim_date:             31 rows (August 2026)
  - kiosko.dim_product:          4 columns, zero of history, P002 historized via time travel
      CORRECT margin (AS OF snap_v1, snacks)      = 10.8
      BROKEN margin (current, health-snacks)      = 9.36
      -- identical to data-modeling-for-analytics-guide M8 and dbt-analytics-engineering-guide M8
  - table.upsert() reproduces, row for row, the same result as table.overwrite()
  - kiosko.fact_orders_at_scale: 10,000,040 rows, revenue 26537606.15, 2 partition schemes coexisting

Seventeen asserts, none decorative: they confirm each table's revenue, the three stores' country, dim_product's grain and schema, P002's two margins against data-modeling-for-analytics-guide's and dbt-analytics-engineering-guide's canonical numbers, the exact equivalence between overwrite() and upsert(), and fact_orders_at_scale's complete final state — six tables visible in the namespace (the lakehouse's five plus lesson 6's check table), all coexisting in the same catalog.

Criterion: when a lakehouse on Iceberg wins, and against what

Lesson 2's brief asked, alongside the assembled lakehouse, for a criterion-driven decision: when is a lakehouse on Iceberg the correct choice, versus a classic managed warehouse (BigQuery, Snowflake) or versus plain Parquet with nothing else? The answer isn't "Iceberg always wins" — it's a table of concrete trade-offs, with this guide's evidence as its foundation:

Dimension	Plain Parquet (foundations, M6)	Managed warehouse (BigQuery/Snowflake)	Lakehouse on Iceberg (this guide)
Write atomicity	No — overwrite-partition leaves a real risk window (module 4, lesson 2)	Yes, managed by the engine, with no user thought needed	Yes, verified with a real CommitFailedException (module 4, lesson 7)
Time travel / history	None — columns have to be designed by hand	Yes, with limited retention and the vendor's own syntax	Yes, native, unlimited snapshots while not expired (module 3)
Compute cost	$0, but with no query engine of its own	Per query or per reserved warehouse — charges even when data doesn't change	$0 in local storage; the query engine (Spark, Trino, DuckDB) is chosen separately
Portability across engines	High — any engine reads Parquet	Low — the data lives inside the vendor's warehouse	High — the same kiosko.dim_product got read by PyIceberg and Spark, with no data duplication (module 6)
Schema evolution with no downtime	No — changing a column means rewriting files	Yes, with the engine's own syntax	Yes, verified without touching a single data file (module 4)
Operational control (catalog, maintenance)	None — it's whoever writes each file's responsibility	None — the vendor manages it, with less visibility	Total, but requires operating it (catalogs, expire_snapshots, module 7)
Learning curve	Low	Low for SQL, high for understanding the real cost	Medium-high — requires understanding snapshots, partition specs, catalogs

This table's honest reading, with this guide's eight modules' evidence: a lakehouse on Iceberg wins when the central problem is data portability across engines (the same Parquet files, read by PyIceberg and by Spark, with nothing duplicated) combined with a real need for time travel and schema evolution with no rewriting — exactly the two problems this ecosystem's four previous guides had already run into without solving from the storage layer. A managed warehouse wins when the team prioritizes zero operation — nobody at Kiosko wants to think about catalogs or expire_snapshots — in exchange for accepting per-query cost and single-vendor dependence. And plain Parquet, with nothing else, is still the correct choice when none of the guarantees this whole guide demonstrated are needed — no history to recover, no expected schema evolution, no need for two different engines to read the same table — the exact same case data-engineering-foundations-guide honestly covered at its gold layer, without pretending it needed more.

Diagram: the complete lakehouse, this module's eight lessons

flowchart TB
    L1["L1: overview,\nthe 5 pieces"] --> L2["L2: the brief,\na single catalog"]
    L2 --> L3["L3: fact_orders,\ndim_store, dim_date"]
    L3 --> L4["L4: dim_product,\ntime travel, 10.8/9.36"]
    L4 --> L5["L5: fact_orders_at_scale,\n10M rows, partitioned"]
    L5 --> L6["L6: table.upsert(),\nsame result"]
    L6 --> L7["L7: 7 sibling guides,\nwhat's left"]
    L7 --> L8["L8 (this project):\nthe 5 tables, one script,\n17 assert"]
    L8 --> END["Whole guide CLOSED:\nM1-M8, en/, 106.15/10.8\nverified end to end"]

Closing the whole guide's promise, module by module

What each module promised	Evidence this guide delivered it
M1: from loose Parquet to a real Iceberg table	kiosko.fact_orders, catalog, schema, snapshot — 40 rows, 106.15
M2: the complete anatomy, catalog → data files	Mapped on disk and with table.inspect, no surprises in this capstone
M3: time travel, zero history columns	dim_product, snap_v1, correct margin 10.8 — reproduced in this module's L4
M4: schema evolution with no rewriting	dim_store with country since this module's first commit — the mechanism already verified in M4
M5: hidden and evolved partitioning	fact_orders_at_scale, 10,000,040 rows, 2 spec_ids coexisting — reproduced in this module's L5
M6: MERGE INTO and native upserts	table.upsert() identical to overwrite(), verified byte for byte in this module's L6
M7: catalogs, maintenance, Delta Lake	Named and contrasted; operational hygiene with real evidence
M8: the complete lakehouse	The five tables, one catalog, 106.15/10.8 verified — this project

No module in this guide is left with an unproven promise. Kiosko's total revenue (106.15) and P002's correct margin (10.8, via time travel) are, at this point, the same numbers seven different guides in the data-engineering-ecosystem already confirmed — data-engineering-foundations-guide, python-for-data-engineering-guide, data-modeling-for-analytics-guide, dbt-analytics-engineering-guide, spark-and-distributed-processing-guide, and now this one — each with a different engine and a different technique, all in agreement.

Common mistakes

Running this project on a catalog that already has tables 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 creation 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_first_lakehouse.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.

Interpreting the criterion table as an automatic formula, with no business context. What happens: someone tries to turn the "when a lakehouse wins" table into a mechanical checklist: counting how many rows favor each option and picking the one with the most votes. Why it happens: a comparison table lends itself to being read as a scoreboard, instead of as a map of trade-offs that depend on each team's real context. How to spot it: if your conclusion is "Iceberg won 5 rows against 2, so it's always the best option," you missed this section's central point — the "learning curve" and "operational control" rows are precisely the real costs a team with little operational capacity shouldn't ignore. How to fix it: use the table to identify which of the seven dimensions really matter in your context — does Kiosko need portability across engines, or just a monthly dashboard? — and let that priority, not a row count, decide.

Assuming kiosko.dim_product_merge_check is a mistake in this project, because it doesn't show up in lesson 2's brief. What happens: someone, on seeing six tables in Step 12/12 instead of five, thinks the script has a bug. Why it happens: the brief and this module's lesson 1 talk about "the lakehouse's five tables," and dim_product_merge_check isn't one of them. How to spot it: check Step 8 — that table exists solely to verify, with isolated evidence, that table.upsert() produces the same result as table.overwrite() on kiosko.dim_product, exactly like this module's lesson 6 did. How to fix it: it isn't a mistake — it's lesson 6's same check table, included in this integrated project because its verification (same_result == True) is one of the seventeen the final assert confirms.

Exercises

Exercise 1 — Run the full project yourself, from scratch. In a new directory, with raw_orders.py and kiosko_scale.py in the same place, run python3 kiosko_first_lakehouse.py. Confirm you see the twelve steps complete and the final success message.

See solution

If raw_orders.py and kiosko_scale.py are in the same directory and PyIceberg is installed, the output should exactly reproduce this lesson's structure: twelve numbered steps, followed by the final verification with the six business confirmations. Step 9 — loading ten million rows — is the slowest; the rest runs in seconds.

Exercise 2 — Break an assert on purpose, and watch it fail. Temporarily change DIM_PRODUCT_V2 so P002 has unit_cost=0.75 instead of 0.68, run the script again, and observe which assert fails first. Then revert the change.

See solution

The first assert to fail should be assert current_p002["category"] == "health-snacks" and current_p002["unit_cost"] == 0.68 — because you changed V2's value, the current state no longer matches what the script expects. If you fixed that assert to accept 0.75, the next one to fail would be assert round(margin_broken["health-snacks"], 2) == 9.36, because the broken margin would change with the new cost. This exercise confirms, once more, that this project's asserts are chained to Kiosko's exact canonical values, verified in cascade.

Exercise 3 — Explain, in your own words, why this lesson's criterion table doesn't include a "query speed" row. In 3-4 sentences, justify why this guide, deliberately, didn't compare query performance among plain Parquet, a managed warehouse, and Iceberg.

See solution

Comparing query speed rigorously would require measuring real execution times, under comparable load conditions, over data volumes representative of production — exactly the kind of analysis this guide's DESIGN doc deliberately delegated to advanced-sql-querying-guide (execution plans, tuning) and to spark-and-distributed-processing-guide (distributed computing in depth). Including a "speed" row with no such rigor — based on impressions instead of controlled measurements — would have violated the same "verify, don't trust" discipline that held up every assert in this whole guide. The criterion table deliberately limits itself to the dimensions this guide did verify with direct evidence: atomicity, time travel, portability, schema evolution, storage cost, and learning curve — it doesn't invent a number it never measured.

Summary and next step: the whole guide's close

With this project you close module 8 — and with it, close lakehouse-and-iceberg-guide in its entirety. You assembled Kiosko's lakehouse's five tables into a single catalog, with seventeen automated asserts confirming: total revenue 106.15, correct country in all three stores, 31 calendar rows, P002's margin recovered with time travel (10.8 correct, 9.36 broken — the same numbers data-modeling-for-analytics-guide and dbt-analytics-engineering-guide already confirmed with different engines), exact equivalence between table.overwrite() and table.upsert(), and 10,000,040 rows of fact_orders_at_scale with two partition schemes coexisting.

By this guide's close, Kiosko has a real lakehouse on Apache Iceberg: verified ACID transactions, automatic snapshots on every commit, time travel with no history column at all, schema and partition evolution without rewriting a single file, and two native merge paths converging on the same result. Every one of those guarantees was, when this guide opened in its module 1, a real ceiling plain-Parquet-as-a-file couldn't solve on its own — the same ceiling data-engineering-foundations-guide, data-modeling-for-analytics-guide, dbt-analytics-engineering-guide, and spark-and-distributed-processing-guide had each already run into, each in its own way, before this guide.

This module's lesson 7 already named, with precise evidence, the seven boundaries this lakehouse leaves pending, and the exact sibling guide that solves each one. This guide doesn't promise to solve them — it promises, and delivers, a real table format, working end to end, on the same Kiosko case six previous guides in the ecosystem already used to verify their own guarantees.

Resources

• PyIceberg — official documentation (quickstart), the complete flow this project integrates: load_catalog(), create_namespace(), create_table(), append(), overwrite(), upsert(), update_spec(). py.iceberg.apache.org. In English.
• PyIceberg — complete API reference. py.iceberg.apache.org/api. In English.
• Apache Iceberg — official documentation, reference version 1.11.0. iceberg.apache.org/docs/latest. In English.
• data-modeling-for-analytics-guide DESIGN doc — source of the canonical 106.15/10.8/9.36 numbers this project verifies with assert, for the second consecutive guide. src/guides/data-modeling-for-analytics-guide/DISENO.md. In Spanish.
• dbt-analytics-engineering-guide DESIGN doc — third independent confirmation of the same numbers. src/guides/dbt-analytics-engineering-guide/DISENO.md. In Spanish.
• spark-and-distributed-processing-guide DESIGN doc — source of the original fact_orders.parquet and of the at-scale dataset this project fully rebuilds. src/guides/spark-and-distributed-processing-guide/DISENO.md. In Spanish.
• This guide's DESIGN doc — the full map of the eight modules, the single source of truth this project closes. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.