Module 5: Hidden Partitioning And Partition Evolution

Project: Kiosko's `fact_orders_at_scale`, partitioned

Description

This project closes module 5. You saw Spark's visible cabinet, with its hand-labeled folders (lesson 2). You asked the same business question without knowing any physical layout (lesson 3). You learned the three partition transforms and when to use each one (lesson 4). You created kiosko.fact_orders_at_scale with a real PartitionSpec, loaded ten million rows, and confirmed the hidden query with pruning evidence (lesson 5). You evolved that spec without rewriting anything (lesson 6). And you saw, with table.inspect.partitions(), how both schemes coexist in the same table (lesson 7). Only one step is left: bringing the seven pieces together in a single script, run end to end, with automated asserts confirming every claim.

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 promise this module opened in lesson 1: a table partitioned at scale, queried without knowing its physical layout, evolved forward without rewriting a single one of the ten million rows that already existed.

An analogy: the carrier, end to end, in a single shift

Lessons 2 through 7 of this module built, one piece at a time, the complete evidence for hidden partitioning and its evolution: the contrast with Spark's visible cabinet (lesson 2), the query that doesn't need to know it (lesson 3), the three transforms' vocabulary (lesson 4), the real table at scale with its measurable pruning (lesson 5), evolution without rewriting (lesson 6), and both schemes coexisting (lesson 7). This project is the carrier's full shift, start to finish, without breaks: he receives ten million pieces, organizes them, answers a query, changes his criterion halfway through, receives more mail, and shows you, at the end of the shift, the complete file — old and new, with no contradiction.

The material: everything this module built, in one place

You need, in a new working directory:

kiosko_fact_orders_at_scale_project/
├── kiosko_scale.py                      (the deterministic generator, identical to the module's)
└── kiosko_partitioned_at_scale.py       (this project: brings the 7 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.fact_orders_at_scale from scratch, so it doesn't depend on any file from this module's previous lessons — only on the kiosko catalog existing in the directory where you run it (if you already have other tables from modules 1 through 4 in that same catalog, this project leaves them untouched).

Scale note. This project loads the full 10,000,000 rows — the same 250,000 franchises from lesson 5, not a reduced sample; on a modern laptop, the full generation and load take roughly half a minute. If your machine is more limited, NUM_FRANCHISES is the only constant you need to reduce — the formula (NUM_FRANCHISES × 40 rows, 106.15 × NUM_FRANCHISES in revenue) holds for any value, exactly as exercise 1 of the spark-and-distributed-processing-guide lesson that originated this dataset demonstrated.

The reference solution, verified

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

KIOSKO_WEEK = [
    {"order_id": "ORD-1001", "store_id": "S01", "product_id": "P001", "quantity": 3, "unit_price": 0.55, "order_ts": "2026-08-03T08:14:00"},
    # ... the full 40 rows of Kiosko's real week
]


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"],
            }

# kiosko_partitioned_at_scale.py -- module 5's closing project
# fact_orders_at_scale partitioned, queried hidden, evolved without rewriting
import os
from datetime import datetime

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

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)


def data_file_count(table) -> int:
    return len(table.inspect.files().to_pylist())


def main() -> None:
    print("=== Kiosko: fact_orders_at_scale partitioned, hidden, evolved ===\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")
    table = catalog.create_table(
        "kiosko.fact_orders_at_scale", schema=FACT_ORDERS_AT_SCALE_SCHEMA, partition_spec=INITIAL_SPEC,
    )
    print(f"Step 1/8 -- catalog ready, kiosko.fact_orders_at_scale created with initial spec:")
    print(f"           {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)
    files_before_evolution = data_file_count(table)
    print(f"Step 2/8 -- {row_count} rows loaded ({NUM_FRANCHISES} franchises x 40), "
          f"total revenue {round(total_revenue, 2)}, {files_before_evolution} data file(s)")

    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/8 -- 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")
    files_after_evolution = data_file_count(table)
    print(f"Step 4/8 -- spec evolved with add_field('order_ts', DayTransform(), 'order_day'). "
          f"Files: {files_before_evolution} -> {files_after_evolution} (unchanged: "
          f"{files_before_evolution == files_after_evolution})")
    print(f"           current spec: {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/8 -- 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
    }
    print(f"Step 6/8 -- table.inspect.partitions() has {len(partitions)} partition rows, "
          f"spec_id present: {spec_ids_present}, rows per spec: {rows_by_spec}")

    row_count_final = table.scan().to_arrow().num_rows
    revenue_final = revenue_of(table.scan().to_arrow())
    print(f"Step 7/8 -- final state: {row_count_final} rows, total revenue {round(revenue_final, 2)}")

    total_snapshots = len(table.history())
    print(f"Step 8/8 -- table.history() has {total_snapshots} snapshots total\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 == 1 and files_touched_all == 3, \
        "the filtered query must touch fewer files than the full scan"
    assert files_before_evolution == files_after_evolution, \
        "evolving the spec must not rewrite any data file"
    assert spec_ids_present == [0, 1], "both partition schemes must coexist in the same table"
    assert rows_by_spec[0] == 10_000_000, "the bulk load's rows must still be under the original spec"
    assert rows_by_spec[1] == 40, "only the new franchise must be under the evolved spec"
    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 loaded under an initial PartitionSpec (IdentityTransform over store_id)")
    print("  - store_id == 'S01' returns 9,575,000.00 with the code never mentioning any folder")
    print("  - the filtered scan touched 1 of 3 files -- hidden partitioning's real pruning")
    print("  - update_spec().add_field(DayTransform) did not rewrite a single existing data file")
    print("  - inspect.partitions() shows the two partition schemes coexisting in the same table")


if __name__ == "__main__":
    main()

What to expect (verified by running the real python3 kiosko_partitioned_at_scale.py, end to end, in a new directory; no snapshot-id gets printed as a literal, following this guide's hard rule):

=== Kiosko: fact_orders_at_scale partitioned, hidden, evolved ===

Step 1/8 -- catalog ready, kiosko.fact_orders_at_scale created with initial spec:
           [
  1000: store_id: identity(3)
]
Step 2/8 -- 10000000 rows loaded (250000 franchises x 40), total revenue 26537500.0, 3 data file(s)
Step 3/8 -- hidden query store_id == 'S01': 4000000 rows, revenue 9575000.0, touched 1/3 files
Step 4/8 -- spec evolved with add_field('order_ts', DayTransform(), 'order_day'). Files: 3 -> 3 (unchanged: True)
           current spec: [
  1000: store_id: identity(3)
  1001: order_day: day(7)
]
Step 5/8 -- franchise 250000 (40 rows) lands under the new spec (with order_day)
Step 6/8 -- table.inspect.partitions() has 23 partition rows, spec_id present: [0, 1], rows per spec: {0: 10000000, 1: 40}
Step 7/8 -- final state: 10000040 rows, total revenue 26537606.15
Step 8/8 -- table.history() has 2 snapshots total

=== Final verification ===

All verifications passed:
  - 10,000,000 rows loaded under an initial PartitionSpec (IdentityTransform over store_id)
  - store_id == 'S01' returns 9,575,000.00 with the code never mentioning any folder
  - the filtered scan touched 1 of 3 files -- hidden partitioning's real pruning
  - update_spec().add_field(DayTransform) did not rewrite a single existing data file
  - inspect.partitions() shows the two partition schemes coexisting in the same table

Notice step 8: table.history() reports two entries, not eight or five, even though this script went through eight distinct narrative stages. The two entries are, in order: the append() of the ten million rows (step 2), and the append() of the new franchise (step 5). Neither the spec evolution (step 4) nor the two read queries (step 3) added any entry to the snapshot history — exactly the same pattern you already saw in module 4's project: metadata operations (schema or partitioning) never create snapshots; only data writes do.

Diagram: where you came from, where you landed

flowchart LR
    A["Modules 1-4:\nfact_orders, dim_product,\ntime travel, evolved dim_store"] --> B["Lesson 2:\nSpark's visible cabinet,\non disk, with real code"]
    B --> C["Lesson 3:\nsame query,\nwithout knowing the layout"]
    C --> D["Lesson 4:\nIdentityTransform,\nBucketTransform, DayTransform"]
    D --> E["Lesson 5:\n10M rows, S01 = 9,575,000.00,\n1 of 3 files touched"]
    E --> F["Lesson 6:\nupdate_spec() adds order_day,\n0 files rewritten"]
    F --> G["Lesson 7:\ninspect.partitions():\nspec_id 0 and 1 coexisting"]
    G --> H["This project:\nthe 7 pieces, one script,\nautomated assert"]
    H --> I["Module 6:\nMERGE INTO and\nnative upserts"]

Closing the module's promise, point by point

What lesson 1 promised	Evidence this module delivered it
Spark's folder-partitioning cost, with real evidence	Lesson 2: real Hive layout on disk, `store_id` disappears from a read that doesn't declare it
The same query, without knowing the layout	Lesson 3: identical `row_filter="store_id == 'S01'"` against a table without partitioning
Precise vocabulary for the three transforms	Lesson 4: `IdentityTransform`, `BucketTransform`, `DayTransform`, run and contrasted
`kiosko.fact_orders_at_scale` partitioned, with `S01 = 9,575,000.00`	Lesson 5 and this project: `10,000,000` rows, real pruning of `1` of `3` files
Spec evolution without rewriting	Lesson 6 and this project: `files_before_evolution == files_after_evolution`, verified with `assert`
Both partition schemes coexisting	Lesson 7 and this project: `spec_ids_present == [0, 1]`, `rows_by_spec == {0: 10_000_000, 1: 40}`

This project didn't touch kiosko.fact_orders, kiosko.dim_product, or kiosko.dim_store — those tables stay exactly as modules 1 through 4 left them. What this project delivers is exactly what it promised: a table partitioned at full scale, queried without knowing its physical layout, evolved forward without rewriting a single one of the ten million rows already written, with the guarantee verified with real code, not just quoted from documentation.

Common mistakes

Running this project on a catalog that already has kiosko.fact_orders_at_scale from an earlier lesson in this module. What happens: someone runs this project in the same directory where they already completed lessons 5 through 7, and catalog.create_table(...) fails because the table is 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_partitioned_at_scale.py, you already have a catalog with kiosko.fact_orders_at_scale registered in the same directory. How to fix it: run this project in a new working directory, separate from where you did lessons 5 through 7 — exactly as this lesson's "The material" suggests.

Expecting step 6 to print record_count adding up to exactly 10,000,040 in a single row. What happens: someone, on seeing rows_by_spec: {0: 10000000, 1: 40}, expects to also find a consolidated row with the combined total (10,000,040) somewhere in table.inspect.partitions()'s output. Why it happens: it's natural to look for a "grand total" in any tabular report, like the TOTAL at the bottom of an invoice. How to spot it: if your code looks for a row with an empty or None spec_id representing the aggregate, revisit this project's step 6 — rows_by_spec gets calculated by adding up in Python, after reading inspect.partitions()'s result, not because the method delivers it that way. How to fix it: table.inspect.partitions() always returns one row per unique combination of spec_id and partition values — never a consolidated total; any additional aggregation you need (like rows_by_spec in this project) is your own code's responsibility, on top of the already-obtained result.

Exercises

Exercise 1 — Run the full project yourself, from scratch. In a new directory, run python3 kiosko_partitioned_at_scale.py. Confirm you see the eight steps complete and the final message with the five verifications.

See solution

If PyIceberg is installed in your environment, the output should exactly reproduce this lesson's structure: eight numbered steps, followed by the final verification with the five success messages. Generating the ten million rows is the script's slowest part — expect something close to half a minute on a modern laptop; the rest of the steps are nearly instant.

Exercise 2 — Break an assert on purpose, and watch it fail. Temporarily change NUM_FRANCHISES from 250_000 to 100_000, run the script again, and observe which assert fails first. Then revert the change.

See solution

The first assert to fail is assert row_count == NUM_FRANCHISES * 40 == 10_000_000 — because with 100_000 franchises, row_count is 4_000_000, which no longer matches the literal 10_000_000 in the second half of that chained comparison. This exercise demonstrates this project's asserts don't only verify internal consistency (row_count == NUM_FRANCHISES * 40, which would hold for any value) — they also verify, with an explicit literal, that this guide's full dataset is specifically the 250,000-franchise one, no more, no less. If you needed to run this project with a different NUM_FRANCHISES due to your machine's limitations, you'd also have to adjust the asserts' literals, exactly as this lesson's scale note suggests.

Exercise 3 — Explain, in your own words, why this project verifies files_touched_s01 == 1 instead of only verifying s01_scan.num_rows == 4_000_000. In 3-4 sentences, justify why the assert on files touched is as important as the assert on the row count.

See solution

Verifying only the row count — that the filtered query returns S01's correct 4,000,000 rows — would confirm the business result is correct, but wouldn't confirm why it's efficient to obtain it. A different implementation, one that scanned all three files in full and discarded the non-S01 rows after reading all of them (the behavior this whole module argues partitioning avoids), could arrive at exactly the same final result, with none of the pruning benefits this module set out to demonstrate. Verifying files_touched_s01 == 1 (against files_touched_all == 3) confirms the module's central claim — that partitioning by store_id lets the engine ignore entire files without opening them — with direct evidence about how many files the execution plan touched, not just about the query's final result. It's the same "verify the path, not just the destination" discipline you already saw in modules 3 and 4's projects.

Summary and next step: this module's close

With this project you close module 5. You integrated the seven previous lessons — the contrast with Spark's visible cabinet, the hidden query, the three partition transforms, the table at scale with real pruning, evolution without rewriting, and both schemes coexisting — into a single script, run end to end, with automated asserts confirming every claim with evidence, not with a promise.

Kiosko has, for the first time in this ecosystem, a table partitioned at real scale whose partition scheme evolved after the data already existed, with that evolution never touching a single already-written Parquet file — and a business query that never, at any point, had to mention how those files are organized internally.

Where you go next. Module 6 — MERGE INTO and native upserts — puts side by side the three ways Kiosko already solved "update P002 without losing its history" — hand-written MERGE INTO in DuckDB (data-modeling), automated dbt snapshot (dbt) — and adds the fourth: Iceberg's native MERGE INTO via Spark SQL, and PyIceberg's table.upsert() as a one-hundred-percent Python alternative. It's also the only module in this guide that needs the JVM — reusing the PySpark that spark-and-distributed-processing-guide already left installed, stated that explicitly since its first lesson.

Resources

PyIceberg — official documentation (quickstart), the full catalog, table, PartitionSpec, append(), and update_spec() flow this project integrates. py.iceberg.apache.org. In English.
PyIceberg — API reference, PartitionSpec/PartitionField, the transforms, table.update_spec(), table.scan(...).plan_files(), table.inspect.partitions(). py.iceberg.apache.org/api. In English.
Apache Iceberg — official documentation, "Partitioning" (hidden partitioning, transforms, partition evolution), the formal foundation for everything this project verifies with assert. iceberg.apache.org/docs/latest/partitioning. In English.
spark-and-distributed-processing-guide DESIGN doc — source of generate_orders_at_scale() and the exact numbers for the at-scale dataset this project rebuilds and loads in full. 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 module 6 that follows. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.

← PreviousReading old and new partition schemes, together Next →Module overview: `MERGE INTO` and native upserts