Partitioning `fact_orders_at_scale`

Description

This is the lesson where everything conceptual from lessons 2 through 4 becomes a real table, at full scale. You're going to create kiosko.fact_orders_at_scale with an initial PartitionSpec — IdentityTransform over store_id, the same criterion Spark already used for its folders — you're going to load Kiosko's real ten million rows at scale, and you're going to confirm two things with executed evidence: that lesson 3's hidden query still gives the correct result (9,575,000.00 for S01), and that this time there really is something to prune — with an exact number of files touched, not just a promise.

Connection to the module. This lesson brings together lesson 2's cabinet, lesson 3's folder-free query, and lesson 4's transform vocabulary, into a single table run end to end. Lessons 6 and 7 are going to build on exactly this same table — they don't recreate it from scratch.

Dataset statement, inherited without regenerating the argument. This lesson's ten million rows are synthetic, generated by generate_orders_at_scale(250_000) — the same deterministic generator, with no random, that spark-and-distributed-processing-guide already justified in its module 4. Real Kiosko — three stores, forty orders in a week — never produces that volume; this guide rebuilds the data, with the same generator, because it needs to really load it into an Iceberg table so partitioning and its evolution have something real to move.

An analogy: the cabinet, rebuilt by the carrier himself

Go back to lesson 1's analogy: this lesson is the moment when the carrier receives, for the first time, a real volume of mail — ten million pieces — and organizes his own bags by his own criterion, with nobody having to tell him anything beyond "organize by recipient." From here on, anyone who asks him for "S01's mail" is going to get a fast, correct answer, without ever having had to learn how the carrier organizes his bags internally.

Worked example: create, load, query, measure

Step 1 — Declare the schema and the initial PartitionSpec

# l5_partitioning_fact_orders_at_scale.py
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 PartitionSpec, PartitionField
from pyiceberg.transforms import IdentityTransform

from kiosko_scale import generate_orders_at_scale

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

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),
)

# the same store_id Spark already used to partition by folder (lesson 2),
# now as an Iceberg PartitionSpec -- partition field_ids start at 1000
# by the format's convention (lesson 4)
INITIAL_SPEC = PartitionSpec(
    PartitionField(source_id=3, field_id=1000, transform=IdentityTransform(), name="store_id"),
)

table = catalog.create_table(
    "kiosko.fact_orders_at_scale",
    schema=FACT_ORDERS_AT_SCALE_SCHEMA,
    partition_spec=INITIAL_SPEC,
)
print("kiosko.fact_orders_at_scale created. Initial PartitionSpec:")
print(table.spec())

What to expect (verified by running the real script):

kiosko.fact_orders_at_scale created. Initial PartitionSpec:
[
  1000: store_id: identity(3)
]

source_id=3 because store_id is the third field of the Schema you just declared above (order_id=1, franchise_id=2, store_id=3); field_id=1000 because this is this table's first partition field (lesson 4). The identity(3) text PyIceberg prints confirms both numbers at a glance: the transform (identity) and the source column (3, that is, store_id).

Step 2 — Generate and load the real ten million rows

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())


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),
])

NUM_FRANCHISES = 250_000
rows = list(generate_orders_at_scale(NUM_FRANCHISES))
for r in rows:
    r["order_ts"] = datetime.fromisoformat(r["order_ts"])
pa_table = pa.Table.from_pylist(rows, schema=PA_SCHEMA)

table.append(pa_table)

# the snapshot-id gets assigned by Iceberg at commit time -- captured in a
# variable, never hardcoded (this guide's hard rule, applied since module 1)
snap_after_bulk_load = table.current_snapshot().snapshot_id
row_count = table.scan().to_arrow().num_rows
print(f"table.scan().to_arrow().num_rows = {row_count}")
assert row_count == NUM_FRANCHISES * 40 == 10_000_000

total_revenue = revenue_of(pa_table)
print(f"total revenue = {round(total_revenue, 2)}")
assert round(total_revenue, 2) == 26_537_500.00

What to expect (verified by running the real script; generating and converting the ten million rows to Arrow takes about 30 seconds on a modern laptop, the append() itself under a second — reference numbers, not a performance measurement to reproduce byte for byte):

table.scan().to_arrow().num_rows = 10000000
total revenue = 26537500.0

Exactly ten million rows, exactly 26,537,500.00 in revenue — the same two numbers spark-and-distributed-processing-guide already verified with its own engine. snap_after_bulk_load stayed captured in a variable, ready for any later use that needs it (this lesson doesn't use it again, but lesson 6 does the equivalent thing with its own snapshot).

Step 3 — The hidden query, at full scale

print("=== The hidden query: filter by store_id, never by folder ===")
s01_scan = table.scan(row_filter="store_id == 'S01'").to_arrow()
s01_revenue = revenue_of(s01_scan)
print(f'table.scan(row_filter="store_id == \'S01\'").to_arrow()')
print(f"  rows: {s01_scan.num_rows}, revenue: {round(s01_revenue, 2)}")
assert s01_scan.num_rows == 4_000_000
assert round(s01_revenue, 2) == 9_575_000.00

What to expect:

=== The hidden query: filter by store_id, never by folder ===
table.scan(row_filter="store_id == 'S01'").to_arrow()
  rows: 4000000, revenue: 9575000.0

9,575,000.00 — the same exact number that appears in this guide's DESIGN doc, identical to what spark-and-distributed-processing-guide gets filtering the same store_id over its own partitioned Parquet. The line of code is, literally, the same one you already used in lesson 3 against kiosko.fact_orders without partitioning — no syntax change at all, just a result that now really does have partitioning underneath.

Step 4 — The pruning evidence: how many files each scan touched

print("=== Pruning evidence: how many data files each scan touched ===")
all_files = list(table.scan().plan_files())
s01_files = list(table.scan(row_filter="store_id == 'S01'").plan_files())
print(f"  full scan (no filter): {len(all_files)} data file(s)")
print(f"  filtered scan (store_id == 'S01'): {len(s01_files)} data file(s)")
assert len(all_files) == 3
assert len(s01_files) == 1

What to expect:

=== Pruning evidence: how many data files each scan touched ===
  full scan (no filter): 3 data file(s)
  filtered scan (store_id == 'S01'): 1 data file(s)

This is the real difference from lesson 3, where both numbers — with filter and without — were 1, because there was nothing to prune. Here, a 10-million-row append() on a table partitioned by store_id produced exactly three data files — one per distinct value of the partition column — and the scan filtered by S01 touched exactly one of those three, ignoring the other two entirely without the code explicitly asking for that. That's hidden partitioning really working: lesson 3's same syntax, now with a measurable benefit.

Diagram: from load to pruning

flowchart TB
    A["generate_orders_at_scale(250_000)\n10,000,000 rows"] --> B["table.append(pa_table)"]
    B --> C["3 data files\n(one per store_id, from the PartitionSpec)"]
    C --> D1["data/store_id=S01/...\n4,000,000 rows"]
    C --> D2["data/store_id=S02/...\n3,250,000 rows"]
    C --> D3["data/store_id=S03/...\n2,750,000 rows"]
    E["table.scan(row_filter=\"store_id == 'S01'\")"] -->|"queries store_id,\nnever mentions a folder"| C
    C -.->|"plan_files() decides,\nonly opens D1"| D1

Common mistakes

Expecting table.append() to produce one file per row, or a single giant file, instead of one per partition value. What happens: someone, on seeing the load produced exactly 3 files for 10,000,000 rows, is surprised — they expected many more (one per internal write batch) or exactly one (everything together, with no real partitioning). Why it happens: without having seen the full mechanism before, it's easy not to anticipate that the number of files from an append() is governed, precisely, by how many distinct partition values show up in the data you're loading. How to spot it: if your data file count doesn't match the number of distinct values in your partitioned column (plus some additional fragmentation on very large loads or multiple append() calls), check what PartitionSpec your table has. How to fix it: with IdentityTransform over a three-value column, and a single append() call with all the data in memory at once, 3 files is exactly what's expected — one per store_id. Repeated loads (several successive append() calls) do produce more files, one per combination of call and partition value — that's precisely the "many small files" problem module 7's compaction is going to solve.

Measuring "faster" by comparing this query's time against lesson 3's, instead of comparing files touched. What happens: someone times both queries — against kiosko.fact_orders without partitioning, and against kiosko.fact_orders_at_scale partitioned — and treats the time difference as the central proof that partitioning "works." Why it happens: a stopwatch is intuitive and easy to use, and the result — faster — confirms the expectation. How to spot it: if your evidence that "partitioning helps" is a number of seconds, instead of a number of files, revisit why this lesson measures with plan_files() instead. How to fix it: an execution time depends on the machine, on the system load at that moment, on the operating system's cache — it isn't reproducible across runs or across people. The number of files plan_files() reports is: 1 out of 3 is a structural fact about the table, verifiable with an assert, with no dependence on any clock.

Exercises

Exercise 1 — Reproduce the full load yourself, and verify the four central numbers. On your own machine, with kiosko.fact_orders_at_scale freshly created, run this lesson's four steps. Confirm 10,000,000 rows, 26,537,500.00 total revenue, 9,575,000.00 for S01, and 1 out of 3 files touched by the filtered scan.

Exercise 2 — Repeat the filtered query for S02 and S03, and confirm the full breakdown. Using step 3's same syntax, filter by store_id == 'S02' and store_id == 'S03', and confirm with assert the revenues 9,700,000.00 and 7,262,500.00 respectively — the same per-store breakdown you already know from module 1.

s02_scan = table.scan(row_filter="store_id == 'S02'").to_arrow()
s03_scan = table.scan(row_filter="store_id == 'S03'").to_arrow()
assert round(revenue_of(s02_scan), 2) == 9_700_000.00
assert round(revenue_of(s03_scan), 2) == 7_262_500.00
print("Verification: full per-store breakdown confirmed with assert")

Exercise 3 — Prediction: how many files would a scan filtered by product_id, instead of store_id, touch? Without running it, predict: if you wrote table.scan(row_filter="product_id == 'P002'").plan_files() against this same table, how many of the three data files do you expect it to touch?

Summary and next step

In this lesson you created kiosko.fact_orders_at_scale with a real PartitionSpec — IdentityTransform over store_id — loaded Kiosko's full ten million rows at scale, and confirmed two facts with assert: the hidden query still returns the correct result (9,575,000.00 for S01, mentioning no folder at all), and this time there really is pruning — one file out of three, not one out of one.

Before moving on you should be able to: explain why the load produced exactly three data files; reproduce the hidden query and its pruning evidence; and anticipate what would happen if you filtered by a column that isn't part of the PartitionSpec (exercise 3).

You have a partitioned, working table. But the partitioning you chose today — only by store_id — doesn't have to be the final one forever. Lesson 6 evolves this same PartitionSpec, adding a second dimension, without rewriting a single one of the ten million rows you already loaded.

Resources

• PyIceberg — API reference, PartitionSpec, PartitionField, catalog.create_table(..., partition_spec=...), table.scan(...).plan_files(). py.iceberg.apache.org/api. In English.
• Apache Iceberg — official documentation, "Partitioning" (hidden partitioning and file pruning as a direct consequence of the PartitionSpec). iceberg.apache.org/docs/latest/partitioning. In English.
• spark-and-distributed-processing-guide DESIGN doc — source of the exact numbers for the at-scale dataset (10,000,000 rows, 26,537,500.00 in revenue, 9,575,000.00/9,700,000.00/7,262,500.00 per store) this lesson verifies with assert. src/guides/spark-and-distributed-processing-guide/DISENO.md. In Spanish.
• This guide's DESIGN doc — module 5's section, with kiosko.fact_orders_at_scale's exact specification and its initial PartitionSpec. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.