Module 1: From File Format To Table Format

Loading Kiosko's fact_orders into Iceberg

Next →

Description

This is the module's central lesson: the kiosko.fact_orders table, empty since lesson 5, finally receives Kiosko's forty real rows for the week. You're going to reconstruct fact_orders.parquet with pyarrow — the same Parquet spark-and-distributed-processing-guide (module 3 of that guide) already left written, reconstructed here so this guide is self-contained — and load it with table.append(), the operation that creates, for the first time, a real snapshot.

Connection to the module. This lesson brings together everything you installed in lessons 4 and 5 — the catalog and the empty table — with the real Kiosko data you already know from the six previous guides. Lesson 7 verifies, with the same number as always (106.15), that the migration was exact.

An analogy: pasting the first photo collection into the album

The album from lesson 5 has a cover and an index, but it's empty. This lesson is the moment to paste, for the first time, a complete photo collection — Kiosko's forty orders for the week — onto its pages. And notice something the analogy predicts precisely: when the librarian finishes pasting this first collection, they don't just update the index to say "there are now forty photos" — they also file, in a separate record, the fact that this was the first collection ever pasted into this album, with its own date and its own reference number. That record is, precisely, what Iceberg calls a snapshot: it isn't a copy of the photos — those are still the same Parquet files as always — it's the archived record of "this is what the whole album looked like, right after this loading event."

Worked example: from fixed data in Python to a real Iceberg table

Step 1 — Kiosko's fixed week, identical to the six previous guides

# raw_orders.py -- Kiosko's fixed week, identical to foundations/data-modeling/dbt/spark
RAW_ORDERS = [
    # Monday 2026-08-03 (8 orders)
    ("ORD-1001", "S01", "P001", 3, 0.55, "2026-08-03T08:14:00"),
    ("ORD-1002", "S01", "P002", 1, 1.20, "2026-08-03T08:20:00"),
    ("ORD-1003", "S02", "P003", 2, 0.75, "2026-08-03T08:31:00"),
    ("ORD-1004", "S01", "P004", 1, 4.50, "2026-08-03T09:02:00"),
    ("ORD-1005", "S03", "P001", 5, 0.55, "2026-08-03T09:15:00"),
    ("ORD-1006", "S02", "P002", 2, 1.20, "2026-08-03T09:47:00"),
    ("ORD-1007", "S03", "P003", 1, 0.75, "2026-08-03T10:05:00"),
    ("ORD-1008", "S01", "P001", 2, 0.55, "2026-08-03T10:22:00"),
    # Tuesday 2026-08-04 (6 orders)
    ("ORD-2001", "S01", "P002", 1, 1.20, "2026-08-04T08:05:00"),
    ("ORD-2002", "S02", "P001", 4, 0.55, "2026-08-04T08:40:00"),
    ("ORD-2003", "S03", "P004", 1, 4.50, "2026-08-04T09:12:00"),
    ("ORD-2004", "S01", "P003", 3, 0.75, "2026-08-04T09:50:00"),
    ("ORD-2005", "S02", "P002", 2, 1.20, "2026-08-04T10:15:00"),
    ("ORD-2006", "S03", "P001", 6, 0.55, "2026-08-04T10:33:00"),
    # Wednesday 2026-08-05 (2 orders)
    ("ORD-3001", "S02", "P004", 2, 4.50, "2026-08-05T08:10:00"),
    ("ORD-3002", "S01", "P001", 1, 0.55, "2026-08-05T08:22:00"),
    # Thursday 2026-08-06 (5 orders)
    ("ORD-4001", "S01", "P001", 4, 0.55, "2026-08-06T08:10:00"),
    ("ORD-4002", "S02", "P003", 2, 0.75, "2026-08-06T08:45:00"),
    ("ORD-4003", "S03", "P002", 1, 1.20, "2026-08-06T09:20:00"),
    ("ORD-4004", "S01", "P004", 1, 4.50, "2026-08-06T09:55:00"),
    ("ORD-4005", "S02", "P001", 3, 0.55, "2026-08-06T10:30:00"),
    # Friday 2026-08-07 (7 orders)
    ("ORD-5001", "S01", "P002", 2, 1.20, "2026-08-07T08:05:00"),
    ("ORD-5002", "S03", "P001", 4, 0.55, "2026-08-07T08:30:00"),
    ("ORD-5003", "S02", "P004", 1, 4.50, "2026-08-07T08:58:00"),
    ("ORD-5004", "S01", "P003", 2, 0.75, "2026-08-07T09:22:00"),
    ("ORD-5005", "S03", "P002", 3, 1.20, "2026-08-07T09:47:00"),
    ("ORD-5006", "S02", "P001", 5, 0.55, "2026-08-07T10:15:00"),
    ("ORD-5007", "S01", "P001", 2, 0.55, "2026-08-07T10:40:00"),
    # Saturday 2026-08-08 (9 orders)
    ("ORD-6001", "S01", "P001", 6, 0.55, "2026-08-08T08:00:00"),
    ("ORD-6002", "S02", "P002", 3, 1.20, "2026-08-08T08:18:00"),
    ("ORD-6003", "S03", "P001", 4, 0.55, "2026-08-08T08:35:00"),
    ("ORD-6004", "S01", "P004", 2, 4.50, "2026-08-08T08:52:00"),
    ("ORD-6005", "S02", "P003", 3, 0.75, "2026-08-08T09:10:00"),
    ("ORD-6006", "S03", "P002", 2, 1.20, "2026-08-08T09:28:00"),
    ("ORD-6007", "S01", "P003", 1, 0.75, "2026-08-08T09:45:00"),
    ("ORD-6008", "S02", "P001", 7, 0.55, "2026-08-08T10:02:00"),
    ("ORD-6009", "S03", "P004", 1, 4.50, "2026-08-08T10:20:00"),
    # Sunday 2026-08-09 (3 orders)
    ("ORD-7001", "S01", "P001", 2, 0.55, "2026-08-09T09:15:00"),
    ("ORD-7002", "S02", "P002", 1, 1.20, "2026-08-09T09:40:00"),
    ("ORD-7003", "S03", "P001", 3, 0.55, "2026-08-09T10:05:00"),
]

Step 2 — Reconstruct fact_orders.parquet with pyarrow

# build_fact_orders_parquet.py
from datetime import datetime

import pyarrow as pa
import pyarrow.parquet as pq

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_ids = {s["store_id"] for s in DIM_STORE}
product_ids = {p["product_id"] for p in DIM_PRODUCT}

order_id_col, store_id_col, product_id_col = [], [], []
quantity_col, unit_price_col, revenue_col, order_ts_col = [], [], [], []

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}")
    order_id_col.append(order_id)
    store_id_col.append(store_id)
    product_id_col.append(product_id)
    quantity_col.append(quantity)
    unit_price_col.append(unit_price)
    revenue_col.append(round(quantity * unit_price, 10))
    order_ts_col.append(datetime.fromisoformat(ts))

fact_orders_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(order_id_col, type=pa.string()),
        pa.array(store_id_col, type=pa.string()),
        pa.array(product_id_col, type=pa.string()),
        pa.array(quantity_col, type=pa.int32()),
        pa.array(unit_price_col, type=pa.float64()),
        pa.array(revenue_col, type=pa.float64()),
        pa.array(order_ts_col, type=pa.timestamp("us")),
    ],
    schema=fact_orders_schema,
)

pq.write_table(pa_table, "fact_orders.parquet")
print(f"fact_orders.parquet written: {pa_table.num_rows} rows, {pa_table.num_columns} columns")

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

fact_orders.parquet written: 40 rows, 7 columns

Step 3 — Load the Parquet into the Iceberg table with table.append()

# load_into_iceberg.py
import os

import pyarrow.parquet as pq
from pyiceberg.catalog import load_catalog

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.load_table("kiosko.fact_orders")

pa_table = pq.read_table("fact_orders.parquet")
print(f"fact_orders.parquet read: {pa_table.num_rows} rows")

table.append(pa_table)

# the snapshot-id is assigned by Iceberg at commit time -- captured
# in a variable, never hardcoded (a hard rule of this guide)
snap_id = table.current_snapshot().snapshot_id
print(f"\nFirst snapshot created, snapshot_id captured in variable: {snap_id}")

scanned = table.scan().to_arrow()
print(f"\nlen(table.scan().to_arrow()) = {scanned.num_rows}")

What to expect (verified by running the actual script; the snapshot_id is a large integer, assigned by Iceberg at the exact moment of the commit — different on every run of yours, never the same twice, so it's shown here as a placeholder instead of a fixed number):

fact_orders.parquet read: 40 rows

First snapshot created, snapshot_id captured in variable: <snapshot-id assigned in your run, different each time>

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

Forty rows read from the Parquet, forty rows confirmed inside the Iceberg table after table.append() — the same number, with no loss and no duplication. And for the first time in this guide, table.current_snapshot() stops being None: a real snapshot exists, with a snapshot_id that your run assigned, different from the one any other run would assign — including a second run of yours, if you deleted the table and recreated it. This is exactly why this guide's rule forbids hardcoding a snapshot-id: it isn't a reproducible number, it's an identifier the commit itself generates, and the code above captures it into the snap_id variable immediately after it's created, instead of assuming what it's going to be.

Diagram: what appeared on disk after the append()

flowchart TB
    A["table.append(pa_table)"] --> B["1 new data file\ndata/00000-0-<uuid>.parquet"]
    A --> C["1 new manifest file\nmetadata/<uuid>-m0.avro\n(lists THAT data file)"]
    A --> D["1 new manifest list\nmetadata/snap-<snapshot_id>-0-<uuid>.avro\n(lists THAT manifest file)"]
    A --> E["1 new metadata.json\nmetadata/00001-<uuid>.metadata.json\n(points to the new snapshot)"]
    E -->|"the catalog now points here"| F["kiosko_catalog.db:\nmetadata_location updated"]

After this lesson, kiosko_warehouse/kiosko/fact_orders/ contains, verified on disk:

kiosko_warehouse/kiosko/fact_orders/
├── data/
│   └── 00000-0-<uuid>.parquet              (the 40 rows, plain Parquet)
└── metadata/
    ├── 00000-<uuid>.metadata.json          (lesson 5: empty table, snapshot=None)
    ├── 00001-<uuid>.metadata.json          (this lesson: points to the first snapshot)
    ├── <uuid>-m0.avro                       (manifest file: lists the data file)
    └── snap-<snapshot_id>-0-<uuid>.avro     (manifest list: lists the manifest file)

You don't need to understand yet, in detail, what a manifest file or a manifest list is — the entire module 2 of this guide is dedicated to that exact chain (catalog → metadata → manifest list → manifest files → data files), opening each of these files with the PyIceberg API. What's worth confirming here is something simpler and more important: there are two .metadata.json files, not just one. The first (00000-...) is the one lesson 5 created, with the empty table. The second (00001-...) is the one this lesson just created — and the first one still exists, without having been deleted or overwritten. That is, in the most concrete practical terms possible, the "every write is a new snapshot, nothing gets overwritten" guarantee you're going to explore in depth in module 3.

Common mistakes

Trying table.append() with a pyarrow.Table whose schema doesn't match exactly. What happens: someone builds their own pa.Table with, for example, quantity as int64 instead of int32, or with the columns in a different order than Iceberg's schema, and table.append() fails with a schema validation error. Why it happens: pyarrow is flexible about numeric types by default (int64 is the type it uses if nothing is specified), and it's easy to miss that the schema declared in lesson 5 asked for int32 for quantity. How to spot it: if table.append() fails mentioning a type or schema mismatch, compare, column by column, the pa_table.schema you're passing against the Iceberg table's table.schema(). How to fix it: this lesson's build_fact_orders_parquet.py declares the pa.schema() explicitly, column by column, with the exact same types as lesson 5's Iceberg Schema — replicating that discipline avoids this error entirely.

Running load_into_iceberg.py twice, and being surprised to see 80 rows instead of 40. What happens: someone runs step 3's script once, sees 40 rows, runs it again out of curiosity (or by mistake), and now table.scan().to_arrow().num_rows gives 80. Why it happens: table.append() is, exactly as its name says, an operation that adds rows — it doesn't replace the table's content. Every call creates a new snapshot with the new rows added to whatever was already there, the same behavior (by design) as a normal SQL INSERT. How to spot it: if your table's row count is a multiple of 40, check how many times you ran table.append() against the same table. How to fix it: if you need to start over from scratch, delete the whole table with catalog.drop_table("kiosko.fact_orders") and recreate it starting from lesson 5 — or, later in this guide (module 3), use table.overwrite() instead of table.append() when the intent is to replace the content, not add to it.

Exercises

Exercise 1 — Reproduce the full load yourself. On your own machine, with lesson 5's empty kiosko.fact_orders table still available, run this lesson's three steps in order: reconstruct fact_orders.parquet, load it with table.append(), and confirm len(table.scan().to_arrow()) == 40. Write down the snapshot_id your own run assigned.

See solution

If you followed the three steps exactly, you should see fact_orders.parquet written: 40 rows, 7 columns, followed by fact_orders.parquet read: 40 rows and len(table.scan().to_arrow()) = 40. Your snapshot_id is going to be a large integer, different from anyone else's doing this same exercise — and also different from any earlier run of yours on a recreated table — exactly as this lesson warns: there's no "correct" value it has to match, the only real verification is that the number exists (isn't None) and that the row count is 40.

Exercise 2 — Verify that the two metadata files coexist. Using ls kiosko_warehouse/kiosko/fact_orders/metadata/ from your terminal (not from Python), confirm that exactly two .metadata.json files exist after this lesson, and explain in 1-2 sentences why the first one didn't disappear.

See solution

You should see two files, something like 00000-<uuid>.metadata.json and 00001-<uuid>.metadata.json (the exact UUIDs are going to be different on your run). The first one didn't disappear because Iceberg never overwrites an existing metadata file — every change in the table's state (creating it empty, adding data to it, and any future operation) writes a new metadata file, and the catalog simply updates its pointer toward the most recent one. The old files stay on disk, available, until an explicit maintenance operation (expire_snapshots, which you're going to see in module 7) decides to clean them up.

Exercise 3 — Prediction: what would happen if raw_orders.py had an invalid store_id? Without running it yet, predict: if one of the 40 rows in RAW_ORDERS had "S99" as its store_id (a value that doesn't exist in DIM_STORE), at which exact step of this lesson's worked example do you expect the error to show up?

See solution

The error would show up in step 2 — build_fact_orders_parquet.py — specifically at the if store_id not in store_ids: raise ValueError(...) line, well before the data even tries to be written as Parquet, and well before Iceberg comes into play at all. This is a referential-integrity check done on purpose, in plain Python, replicating the same discipline as the validate_orders() you already saw in data-engineering-foundations-guide — Iceberg guarantees atomicity and structure for the write, but it doesn't, on its own, validate that a store_id corresponds to a real store in DIM_STORE; that's still the responsibility of the code that prepares the data before it reaches table.append().

Summary and next step

In this lesson you reconstructed fact_orders.parquet with pyarrow — the same forty rows as always, the same fixed week from August 3 to 9, 2026 — and loaded it into kiosko.fact_orders with table.append(). You confirmed, with table.current_snapshot() no longer being None, that this table's first real snapshot got created, and you saw on disk that a data file, a manifest file, a manifest list, and a second metadata file appeared — without the first one, from the empty table, disappearing.

Before moving on you should be able to: explain the difference between table.append() and an operation that replaces content; reproduce the full load on your own machine; and explain why the snapshot_id is never hardcoded in this guide's code.

You have forty rows loaded, but you haven't yet verified the number that really matters: whether the total revenue is still 106.15, exactly as in the six previous guides. That is, precisely, lesson 7's job.

Resources

  • PyIceberg — API reference, the exact syntax of table.append() and table.current_snapshot(). py.iceberg.apache.org/api. In English.
  • Apache Iceberg — official documentation, "Table Spec," the formal definition of snapshot, manifest list, and manifest file that module 2 of this guide inspects in depth. iceberg.apache.org/spec. In English.
  • spark-and-distributed-processing-guide DESIGN doc — source of fact_orders.parquet, reconstructed in this lesson with pyarrow in a self-contained way. src/guides/spark-and-distributed-processing-guide/DISENO.md. In Spanish.
  • This guide's DESIGN doc — the hard rule about never hardcoding a snapshot-id, applied in this lesson for the first time. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.
← PreviousCreating your first namespace and tableNext →Verifying the same 106.15 total