Evolving and partitioning at scale

Description

This lesson adds Kiosko's lakehouse's fifth and final table: kiosko.fact_orders_at_scale, with ten million rows, partitioned by store_id with hidden partitioning, and evolved forward with DayTransform over order_ts — exactly the same mechanism this guide's module 5 already demonstrated, now run inside this capstone's single catalog, alongside the other four tables lessons 3 and 4 already loaded.

Connection to the module. This lesson answers lesson 2's brief's third requirement: "partitioning and evolving without rewriting." Unlike kiosko.dim_product in lesson 4, kiosko.fact_orders_at_scale has no business relationship at all with P002's change — it's an independent thread, inherited from spark-and-distributed-processing-guide, that this capstone integrates into the same catalog simply because it is, just like the other four, a real table in Kiosko's lakehouse.

An analogy: the fifth piece, built at the same pace as the other four

If lessons 3 and 4 built the model's foundation, structure, and interior partitions, this lesson builds the building's new wing — one built at a different scale (ten million rows against just forty), with its own internal organization system (hidden partitioning), but anchored to the same ground: the kiosko catalog already holding up the other four pieces. The new wing doesn't need to "talk" to the other four in the sense of sharing data — in fact, it shares no row at all with fact_orders, dim_store, dim_date, or dim_product — it needs to be, physically, part of the same building: the same kiosko_catalog.db, the same kiosko namespace.

Worked example: fact_orders_at_scale, partitioned and evolved, in the same catalog

Step 1 — The deterministic generator, identical to module 5's

# kiosko_scale.py -- the deterministic generator, identical to module 5's
from typing import Any, Dict, Iterator

from raw_orders import RAW_ORDERS

KIOSKO_WEEK = [
    {"order_id": order_id, "store_id": store_id, "product_id": product_id,
     "quantity": quantity, "unit_price": unit_price, "order_ts": ts}
    for order_id, store_id, product_id, quantity, unit_price, ts in RAW_ORDERS
]


def generate_orders_at_scale(num_franchises: int) -> Iterator[Dict[str, Any]]:
    for franchise_id in range(num_franchises):
        for row in KIOSKO_WEEK:
            yield {
                "order_id": f"F{franchise_id:06d}-{row['order_id']}",
                "franchise_id": franchise_id,
                "store_id": row["store_id"],
                "product_id": row["product_id"],
                "quantity": row["quantity"],
                "unit_price": row["unit_price"],
                "order_ts": row["order_ts"],
            }

Step 2 — The schema, the initial spec, and the full 250,000-franchise load

# kiosko_fact_orders_at_scale.py -- module 8, lesson 5
import os
from datetime import datetime

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 DoubleType, IntegerType, NestedField, StringType, TimestampType

from kiosko_scale import generate_orders_at_scale

NUM_FRANCHISES = 250_000

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"),
)
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 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=PA_SCHEMA)

Notice the PartitionField's field_id=1000: the same high range, separated from the columns' field_ids 1-7, module 5 already used — it avoids any collision between a column's identifier and a partition field's.

Step 3 — The complete flow: load, hidden query, evolution, verification

def main() -> None:
    print("=== Kiosko: fact_orders_at_scale, partitioned and evolved ===\n")

    warehouse_path = os.path.abspath("kiosko_warehouse")
    catalog_db_path = os.path.abspath("kiosko_catalog.db")
    catalog = load_catalog(
        "kiosko", type="sql",
        uri=f"sqlite:///{catalog_db_path}", warehouse=f"file://{warehouse_path}",
    )
    table = catalog.create_table(
        "kiosko.fact_orders_at_scale", schema=FACT_ORDERS_AT_SCALE_SCHEMA, partition_spec=INITIAL_SPEC,
    )
    print(f"Step 1/6 -- kiosko.fact_orders_at_scale created with initial spec: {table.spec()}")

    bulk_pa_table = rows_to_pa_table(NUM_FRANCHISES)
    table.append(bulk_pa_table)
    row_count = table.scan().to_arrow().num_rows
    total_revenue = revenue_of(bulk_pa_table)
    print(f"Step 2/6 -- {row_count} rows loaded ({NUM_FRANCHISES} franchises x 40), "
          f"total revenue {round(total_revenue, 2)}")

    s01_scan = table.scan(row_filter="store_id == 'S01'").to_arrow()
    s01_revenue = revenue_of(s01_scan)
    files_touched_s01 = len(list(table.scan(row_filter="store_id == 'S01'").plan_files()))
    files_touched_all = len(list(table.scan().plan_files()))
    print(f"Step 3/6 -- hidden query store_id == 'S01': {s01_scan.num_rows} rows, "
          f"revenue {round(s01_revenue, 2)}, touched {files_touched_s01}/{files_touched_all} files")

    with table.update_spec() as update:
        update.add_field("order_ts", DayTransform(), "order_day")
    print(f"Step 4/6 -- spec evolved: {table.spec()}")

    new_franchise_pa_table = rows_to_pa_table(1, franchise_offset=NUM_FRANCHISES)
    table.append(new_franchise_pa_table)
    print(f"Step 5/6 -- franchise {NUM_FRANCHISES} ({new_franchise_pa_table.num_rows} rows) "
          f"lands under the new spec (with order_day)")

    partitions = table.inspect.partitions().to_pylist()
    spec_ids_present = sorted({p["spec_id"] for p in partitions})
    rows_by_spec = {
        spec_id: sum(p["record_count"] for p in partitions if p["spec_id"] == spec_id)
        for spec_id in spec_ids_present
    }
    row_count_final = table.scan().to_arrow().num_rows
    revenue_final = revenue_of(table.scan().to_arrow())
    print(f"Step 6/6 -- final state: {row_count_final} rows, revenue {round(revenue_final, 2)}, "
          f"spec_id present: {spec_ids_present}, rows per spec: {rows_by_spec}\n")

    print("=== Final verification ===\n")
    assert row_count == NUM_FRANCHISES * 40 == 10_000_000
    assert round(total_revenue, 2) == 26_537_500.00
    assert s01_scan.num_rows == 4_000_000
    assert round(s01_revenue, 2) == 9_575_000.00
    assert files_touched_s01 < files_touched_all, "the filtered query must touch fewer files than the full scan"
    assert spec_ids_present == [0, 1], "both partition schemes must coexist in the same table"
    assert rows_by_spec[0] == 10_000_000
    assert rows_by_spec[1] == 40
    assert row_count_final == 10_000_040
    assert round(revenue_final, 2) == 26_537_606.15

    print("All verifications passed:")
    print("  - 10,000,000 rows under IdentityTransform over store_id, S01 = 9,575,000.00, real file pruning")
    print("  - update_spec().add_field(DayTransform) did not rewrite any existing file")
    print("  - inspect.partitions() confirms both partition schemes coexisting (spec_id 0 and 1)")
    print(f"  - final state: 10,000,040 rows, revenue {round(revenue_final, 2)}")


if __name__ == "__main__":
    main()

What to expect (verified by running the real python3 kiosko_fact_orders_at_scale.py, in the same directory as lessons 3 and 4, without deleting kiosko_warehouse/; on a modern laptop, generating and loading the ten million rows takes around half a minute):

=== Kiosko: fact_orders_at_scale, partitioned and evolved ===

Step 1/6 -- kiosko.fact_orders_at_scale created with initial spec: [
  1000: store_id: identity(3)
]
Step 2/6 -- 10000000 rows loaded (250000 franchises x 40), total revenue 26537500.0
Step 3/6 -- hidden query store_id == 'S01': 4000000 rows, revenue 9575000.0, touched 1/3 files
Step 4/6 -- spec evolved: [
  1000: store_id: identity(3)
  1001: order_day: day(7)
]
Step 5/6 -- franchise 250000 (40 rows) lands under the new spec (with order_day)
Step 6/6 -- final state: 10000040 rows, revenue 26537606.15, spec_id present: [0, 1], rows per spec: {0: 10000000, 1: 40}

=== Final verification ===

All verifications passed:
  - 10,000,000 rows under IdentityTransform over store_id, S01 = 9,575,000.00, real file pruning
  - update_spec().add_field(DayTransform) did not rewrite any existing file
  - inspect.partitions() confirms both partition schemes coexisting (spec_id 0 and 1)
  - final state: 10,000,040 rows, revenue 26537606.15

The final revenue (26,537,606.15) is the sum of the ten million original rows (26,537,500.00) plus the new franchise (106.15, Kiosko's same week, once more) — the exact same arithmetic module 5's closing project already verified.

Diagram: five tables, a single catalog

flowchart TB
    subgraph CAT["kiosko_catalog.db -- a single catalog"]
        FO["kiosko.fact_orders\n40 rows (L3)"]
        DS["kiosko.dim_store\n3 rows, with country (L3)"]
        DD["kiosko.dim_date\n31 rows (L3)"]
        DP["kiosko.dim_product\n4 rows, 2 snapshots,\nno history (L4)"]
        FAS["kiosko.fact_orders_at_scale\n10,000,040 rows,\n2 specs (L5)"]
    end
    L3["Lesson 3"] --> FO
    L3 --> DS
    L3 --> DD
    L4["Lesson 4"] --> DP
    L5["Lesson 5 (this one)"] --> FAS

Going deeper: why this table doesn't interact with the other four

Unlike dim_product in lesson 4 — which explicitly joins against fact_orders, dim_store, and dim_date to calculate a margin — kiosko.fact_orders_at_scale shares no row, no key, no calculation with the rest of the lakehouse in this guide. That isn't an oversight — it's a deliberate, inherited decision, straight from spark-and-distributed-processing-guide's own DESIGN doc, which generated this synthetic dataset (franchise_id from 0 to 249,999) to demonstrate volume, not to model a real franchise business with its own dimensional hierarchy. The reason this table lives in the same kiosko catalog, then, isn't "because it relates to the other four" — it's because it is, physically, one more table in Kiosko's same lakehouse, with the same hidden partitioning and evolution discipline the rest of this guide taught. A real production lakehouse almost always has tables that connect to each other (like fact_orders and dim_product) and tables that coexist in the same catalog with no direct relationship (like this one) — both cases are a normal part of operating a shared catalog.

Common mistakes

Expecting kiosko.fact_orders_at_scale to share some store_id or calculation with kiosko.dim_store. What happens: someone tries to write a JOIN between fact_orders_at_scale and dim_store expecting to enrich the at-scale rows with the country lesson 3 populated. Why it happens: both tables share the store_id column, and it's reasonable to assume any shared store_id implies a business relationship meant for joining. How to spot it: if your JOIN runs with no error but the result doesn't appear in any assert from this lesson or from any previous project, you didn't break anything — you're simply exploring a combination this guide never verified or promised. How to fix it: store_id in fact_orders_at_scale does use the same values (S01, S02, S03) as dim_store — by design, both tables describe Kiosko's same three stores — so that JOIN is technically valid and even interesting as its own exercise; just be clear that no result from that combination is part of this capstone's canonical verified numbers.

Running this lesson without having run lessons 3 and 4 first in the same directory. What happens: someone runs kiosko_fact_orders_at_scale.py in a new directory, with no prior kiosko_warehouse/ or kiosko_catalog.db. Why it happens: unlike previous modules' closing projects — which were self-contained, designed to run alone — this module's lessons 3 through 6 deliberately share the same catalog. How to spot it: if you run this lesson alone, it doesn't actually fail — catalog.create_table("kiosko.fact_orders_at_scale", ...) creates the catalog and namespace if they don't exist — but you'd end up with a catalog holding only this table, not the complete five-table lakehouse this module describes. How to fix it: to reproduce the complete lakehouse as this guide describes it, run lessons 3, 4, and 5 in order, in the same working directory, without deleting anything between them.

Exercises

Exercise 1 — Run the script yourself, in the same directory as lessons 3 and 4. Confirm you see the six steps complete and the final message with the five verifications. Be patient with Step 2 — it's this whole module's slowest step.

See solution

If you ran lessons 3 and 4 first, in the same directory, the output should exactly reproduce this lesson's structure: six numbered steps, followed by the final verification with 10,000,040 rows and 26,537,606.15 in revenue. Step 2 — generating and loading ten million rows — is, by far, the slowest of this module's eight lessons; the rest run almost instantly.

Exercise 2 — Calculate, without running code, how many data files you'd expect to see under spec_id=0 after Step 2, and compare it against what you saw in module 5. Use files_touched_all from Step 3 as a hint.

See solution

files_touched_all in Step 3 is 3 — the same number module 5's closing project already confirmed: one data file per distinct store_id value (S01, S02, S03), a direct consequence of IdentityTransform over store_id physically grouping every row from the same store into the same file, within the same append(). That number doesn't change in this module because the partitioning mechanism — IdentityTransform over a low-cardinality column, three possible values — is exactly the same one module 5 already demonstrated; this capstone reproduces it, it doesn't modify it.

Exercise 3 — Explain, in your own words, why this lesson doesn't need to capture any snapshot_id in a variable, unlike lesson 4. In 2-3 sentences, justify the difference between the two lessons.

See solution

Lesson 4 needs to capture snap_v1 because its central goal is recovering a previous state with time travel — without the captured variable, there'd be no way to ask Iceberg "show me what dim_product looked like before the change." This lesson, instead, never needs to go back to a previous state of fact_orders_at_scale: partition evolution (update_spec()) is a forward operation, that deletes or hides no existing data at all, and this lesson's final assert verifies the table's current state (row_count_final, revenue_final), not a specific past snapshot. That's why table.history() and snapshot_id don't show up in this lesson's main flow, even though they still exist — any intermediate snapshot of this table is still recoverable, this lesson simply doesn't need to ask for it.

Summary and next step

In this lesson you loaded Kiosko's lakehouse's fifth and final table: kiosko.fact_orders_at_scale, with ten million rows partitioned by store_id, evolved with DayTransform over order_ts, and with a new franchise landing under the evolved spec — 10,000,040 final rows, 26,537,606.15 in revenue, two partition schemes coexisting in the same table. Kiosko's lakehouse now has all five of its tables, in the same catalog.

Before moving on you should be able to: explain why fact_orders_at_scale shares no business calculation with the lakehouse's other four tables, despite living in the same catalog; and calculate from memory why the final revenue is 26,537,500.00 + 106.15.

Lesson 6 returns to kiosko.dim_product — not to change it again, but to demonstrate that P002's same change lesson 4 applied with table.overwrite() produces, with table.upsert(), exactly the same business result, through a different path: the format's native route.

Resources

• PyIceberg — official documentation (quickstart), the PartitionSpec, append(), and update_spec() flow this lesson integrates. py.iceberg.apache.org. In English.
• PyIceberg — API reference, the transforms (IdentityTransform, DayTransform), table.scan(...).plan_files(), table.inspect.partitions(). py.iceberg.apache.org/api. In English.
• This same guide, module 5, lesson 8 — source of the original script this lesson reuses inside the capstone's single catalog. ../module-05-hidden-partitioning-and-partition-evolution/en/08-project-kioskos-partitioned-at-scale.md. In English.
• spark-and-distributed-processing-guide DESIGN doc — source of generate_orders_at_scale() and the exact numbers for the at-scale dataset. src/guides/spark-and-distributed-processing-guide/DISENO.md. In Spanish.
• This guide's DESIGN doc — the full map of the eight modules, including the lesson 6 that follows. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.