Module 8: Project Kioskos Lakehouse

Merging updates the native way

Next →

Description

This lesson doesn't change kiosko.dim_product again — it already ended up, since lesson 4, with P002 at health-snacks/0.68, and that state isn't touched. What this lesson does is ask a different question, with new evidence: if Kiosko had to apply this same kind of change routinely — not as an overwrite() that demands the complete table, but as a partial update, "only what changed" — would the business result be the same? The answer, verified with table.upsert() on a separate check table, is yes — byte for byte, row for row, identical to the result table.overwrite() left in lesson 4.

Connection to the module. This lesson revisits, applied inside the assembled lakehouse, the mechanism this guide's module 6 already taught in depth: table.upsert() as Iceberg's native route for applying partial changes, versus overwrite(), which demands recalculating the complete table, and versus Spark SQL's MERGE INTO, which that same module documented as representative because of the real incompatibility between iceberg-spark-runtime-4.0 and Spark 4.1+ (apache/iceberg#15238).

An analogy: the same renovation, with two different contractors, the same final result

Imagine two different contractors each get the job of changing a single store's sign inside a four-unit shopping mall. The first contractor — the one who already worked in lesson 4 — solves the job by rebuilding the mall's entire facade: they take all four signs, change the one that needs it, and reinstall all four together. The second contractor — this lesson's — solves the same job differently: they climb a ladder, remove only the sign that changed, and replace it, without touching the other three. If both did their job right, a visitor arriving afterward can't tell which contractor was there — the mall looks identical. The difference between the two isn't in the final result, it's in how much work and how much information each one needed: the first needed to know, and hand back, all four complete signs; the second only needed the sign that changed.

Worked example: P002's same change, applied with table.upsert()

Step 1 — A check table, separate from the real one

This lesson does not touch kiosko.dim_product again — it creates a new table, kiosko.dim_product_merge_check, loaded with the same V1, to demonstrate the mechanism without risking the state lesson 4 already left verified:

# kiosko_native_merge_check.py -- module 8, lesson 6
import os

import pyarrow as pa
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import DoubleType, NestedField, StringType

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

Step 2 — The key difference: what input each mechanism needs

def main() -> None:
    print("=== Kiosko: P002's same change, applied with table.upsert() -- the native way ===\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}",
    )
    real_dim_product = catalog.load_table("kiosko.dim_product")
    real_rows = sorted(real_dim_product.scan().to_arrow().to_pylist(), key=lambda r: r["product_id"])
    print(f"Step 1/5 -- kiosko.dim_product (lesson 4, changed with overwrite()) is still as it was left: "
          f"P002={next(r for r in real_rows if r['product_id'] == 'P002')['category']}")

    check_table = catalog.create_table("kiosko.dim_product_merge_check", schema=DIM_PRODUCT_SCHEMA)
    check_table.append(pa.Table.from_pylist(DIM_PRODUCT_V1, schema=PA_SCHEMA))
    print("Step 2/5 -- kiosko.dim_product_merge_check created separately, loaded with the same V1")

    only_p002_change = [r for r in DIM_PRODUCT_V2 if r["product_id"] == "P002"]
    result = check_table.upsert(pa.Table.from_pylist(only_p002_change, schema=PA_SCHEMA), join_cols=["product_id"])
    print(f"Step 3/5 -- table.upsert() with ONLY the row that changed: rows_updated={result.rows_updated}, "
          f"rows_inserted={result.rows_inserted} -- nobody calculated the delta by hand")

    check_rows = sorted(check_table.scan().to_arrow().to_pylist(), key=lambda r: r["product_id"])
    ops = [row["operation"] for row in check_table.inspect.snapshots().select(["operation"]).to_pylist()]
    print(f"Step 4/5 -- check table's table.history(): {ops} -- the upsert resolves, "
          f"internally, as overwrite (partial) + append (0 rows, no new row to insert), "
          f"never as the overwrite() of the TOTAL row count that doing it by hand would require")

    same_result = real_rows == check_rows
    print(f"Step 5/5 -- kiosko.dim_product (overwrite, lesson 4) == kiosko.dim_product_merge_check "
          f"(upsert, this lesson), row for row: {same_result}\n")

    print("=== Final verification ===\n")
    assert result.rows_updated == 1 and result.rows_inserted == 0
    assert ops == ["append", "overwrite", "append"], "V1 (append) + upsert (partial overwrite + empty append)"
    assert same_result, "the business result must be identical no matter the write mechanism"
    p002_check = next(r for r in check_rows if r["product_id"] == "P002")
    assert p002_check["category"] == "health-snacks" and p002_check["unit_cost"] == 0.68

    print("All verifications passed:")
    print("  - table.upsert() detected and applied only P002's real change, without touching P001/P003/P004")
    print("  - the business result is identical, row for row, to what table.overwrite() left in lesson 4")
    print("  - the difference is the INPUT each mechanism needs: overwrite() demands the COMPLETE table, "
          "upsert() only demands the DELTA -- a single product, a single row")


if __name__ == "__main__":
    main()

What to expect (verified by running the real python3 kiosko_native_merge_check.py, in the same directory as lessons 3 through 5, without deleting kiosko_warehouse/):

=== Kiosko: P002's same change, applied with table.upsert() -- the native way ===

Step 1/5 -- kiosko.dim_product (lesson 4, changed with overwrite()) is still as it was left: P002=health-snacks
Step 2/5 -- kiosko.dim_product_merge_check created separately, loaded with the same V1
Step 3/5 -- table.upsert() with ONLY the row that changed: rows_updated=1, rows_inserted=0 -- nobody calculated the delta by hand
Step 4/5 -- check table's table.history(): ['append', 'overwrite', 'append'] -- the upsert resolves, internally, as overwrite (partial) + append (0 rows, no new row to insert), never as the overwrite() of the TOTAL row count that doing it by hand would require
Step 5/5 -- kiosko.dim_product (overwrite, lesson 4) == kiosko.dim_product_merge_check (upsert, this lesson), row for row: True

=== Final verification ===

All verifications passed:
  - table.upsert() detected and applied only P002's real change, without touching P001/P003/P004
  - the business result is identical, row for row, to what table.overwrite() left in lesson 4
  - the difference is the INPUT each mechanism needs: overwrite() demands the COMPLETE table, upsert() only demands the DELTA -- a single product, a single row

Notice Step 3: only_p002_change contains a single row, not all four. table.upsert() doesn't need you to hand it P001, P003, or P004 to leave them untouched — it detects them as "no real change matches" and leaves them exactly as they were. table.overwrite(), instead — the mechanism lesson 4 used — always demands receiving the complete table: if you had passed it only P002's row, the other three would have disappeared from the table.

The complete reference — the representative part: MERGE INTO via Spark

For this lesson to document the complete "native way" mechanism module 6 taught, not just the half that runs in this environment, here's MERGE INTO's exact syntax, applied to P002's same change, explicitly marked for what it is — the same real incompatibility (apache/iceberg#15238) module 6 documented:

-- What to expect (representative) -- syntax identical to module 6's, not executed
-- in this environment because of the real incompatibility between iceberg-spark-runtime-4.0 and Spark 4.1+.

MERGE INTO local.kiosko.dim_product AS target
USING local.kiosko.dim_product_staging AS source
ON target.product_id = source.product_id
WHEN MATCHED THEN UPDATE SET
    target.product_name = source.product_name,
    target.category     = source.category,
    target.unit_cost    = source.unit_cost
WHEN NOT MATCHED THEN INSERT (product_id, product_name, category, unit_cost)
VALUES (source.product_id, source.product_name, source.category, source.unit_cost);

What to expect (representative): exactly the same business result table.overwrite() (lesson 4) and table.upsert() (this lesson) already confirmed — P002 with category='health-snacks', unit_cost=0.68, P001/P003/P004 intact. The difference, documented in depth in module 6: this MERGE INTO runs distributed, on Spark, against the local catalog (Hadoop) — a catalog physically distinct from the kiosko (SQL/SQLite) the rest of this capstone uses — and needs the JVM; table.upsert() runs in a single Python process, with neither.

Diagram: three paths, one destination

flowchart TB
    START["P002's change:\nsnacks/0.60 -> health-snacks/0.68"]
    START --> OW["table.overwrite()\n(lesson 4)\ndemands the COMPLETE table"]
    START --> UP["table.upsert()\n(this lesson)\nonly demands the DELTA"]
    START --> MI["MERGE INTO Spark SQL\n(module 6, representative)\ndifferent catalog, needs the JVM"]
    OW --> RESULT["kiosko.dim_product:\nP002 = health-snacks/0.68\nP001/P003/P004 intact"]
    UP --> RESULT
    MI -.->|"same expected result,\nnot executed in this environment"| RESULT

Going deeper: why "the same result" doesn't mean "the same cost"

This lesson's Step 5 confirms overwrite() and upsert() produce exactly the same final state — but that equality hides a real cost difference module 6 already quantified in its lesson 7 ("Choosing between SQL MERGE and Python upsert"): on a four-row table, like dim_product, the difference between "rewriting everything" and "rewriting only the delta" is invisible — both mechanisms finish in milliseconds. But the same decision, applied to a table with millions of rows and only a handful of real changes per day — the kind of table a real production lakehouse handles — stops being an aesthetic choice: overwrite() would force reading, processing, and rewriting every row to change a handful; upsert() processes only the delta. This lesson demonstrates result equivalence at small scale, precisely because at this scale it's easy to verify, row for row, that no mechanism introduced an error — the decision of which to use in production depends on volume, not on which "looks simpler" in a four-row example.

Common mistakes

Modifying kiosko.dim_product (the real table) instead of kiosko.dim_product_merge_check. What happens: someone, when adapting this code, calls catalog.load_table("kiosko.dim_product") instead of creating the separate check table, and runs upsert() on the real table lesson 4 already left verified. Why it happens: it's shorter to write "the real table" than to create a new table just for a check. How to spot it: if this lesson's assert p002_check["category"] == "health-snacks" fails with a table-not-found error, or if a later lesson's kiosko.dim_product shows a snapshot history different from what lesson 4 left, you mixed up the two tables. How to fix it: keep kiosko.dim_product_merge_check as a completely separate table — its only purpose is demonstrating result equivalence, never replacing the state lesson 4 already verified and lesson 8 is going to take for granted.

Assuming ops == ["append", "overwrite", "append"] means upsert() "actually does a complete overwrite internally." What happens: someone, seeing the word "overwrite" in the operations list, concludes table.upsert() isn't different from table.overwrite() at the internal level, and that this whole lesson's comparison is a difference in name only. Why it happens: PyIceberg reuses the same word "overwrite" to describe the snapshot operation type, whether it's called explicitly with table.overwrite() or is the internal result of a partial upsert(). How to spot it: if your conclusion is that both mechanisms read and rewrite the same volume of data, check what pyarrow.Table each one received as an argument — table.overwrite() in lesson 4 received all four rows of V2; table.upsert() in this lesson received a single row. How to fix it: the word "overwrite" in table.history() describes the snapshot type (it replaces existing files instead of only adding), not the volume of data involved — the real difference between the two mechanisms is in how many rows each one had to receive as input, not in how the resulting operation gets labeled.

Exercises

Exercise 1 — Run the script yourself, in the same directory as lessons 3 through 5. Confirm you see the five steps complete and the final message with the four verifications.

See solution

If you ran lessons 3, 4, and 5 first, in the same directory, the output should exactly reproduce this lesson's structure: five numbered steps, followed by the final verification confirming rows_updated=1, rows_inserted=0, and same_result=True. If kiosko.dim_product doesn't exist yet, you ran this lesson before lesson 4.

Exercise 2 — Modify the script to pass table.upsert() all four rows of DIM_PRODUCT_V2, instead of only P002's, and observe whether the result changes. Run the modified script and compare rows_updated/rows_inserted against the original.

See solution

With V2's four rows as the argument, result.rows_updated is still 1 (only P002 really changed) and result.rows_inserted is still 0 — the business result is identical. This exercise confirms something important about table.upsert(): no matter how many rows you hand it, it always detects which ones really changed by comparing against the current state, row by row. Passing it only the delta (like the original script does) has a practical advantage that isn't about correctness — both ways reach the same result — it's about efficiency: whoever writes the pipeline doesn't need to know in advance which row changed, but there's also no need to waste work resending the ones that didn't change if they're already known.

Exercise 3 — Explain, in your own words, why this lesson demonstrates result equivalence with a separate check table, instead of simply re-verifying kiosko.dim_product with an additional assert. In 2-3 sentences, justify this design decision.

See solution

If this lesson ran upsert() directly on kiosko.dim_product, the final result would still be health-snacks/0.68 — but there'd be no way left to distinguish "this ended up this way because of lesson 4's overwrite()" from "this ended up this way because of this lesson's upsert()," because both would produce the same state on the same table. Creating kiosko.dim_product_merge_check as a separate table, loaded independently with the same V1, lets you compare the two mechanisms in isolation, row for row, with evidence they reach the same result through different paths — the same "verify the path, not just the destination" discipline modules 4 and 5's closing projects already applied in this guide.

Summary and next step

In this lesson you applied P002's same change with table.upsert(), on a separate check table, and confirmed — with assert, row for row — the result is identical to what table.overwrite() left in lesson 4. Alongside that evidence, you documented Spark's MERGE INTO's complete syntax, which module 6 already marked as representative because of the real incompatibility between iceberg-spark-runtime-4.0 and Spark 4.1+. Kiosko's lakehouse now has all five of its tables complete, and confirmation that two different write paths — complete overwrite(), partial upsert() — converge on the same business result.

Before moving on you should be able to: explain the input difference between table.overwrite() and table.upsert(), and why that difference matters more at scale than on a four-row table; and name the physically distinct catalog Spark's MERGE INTO would need to run.

Lesson 7 closes the complete ecosystem: it names, one by one, data-engineering-ecosystem's seven sibling guides and what each one solves from what this lakehouse leaves pending.

Resources

  • PyIceberg — official documentation (quickstart), the append() and upsert() flow this lesson integrates. py.iceberg.apache.org. In English.
  • PyIceberg — API reference, table.upsert(), UpsertResult, table.inspect.snapshots(). py.iceberg.apache.org/api. In English.
  • Apache Iceberg — official documentation, "Spark Writes," MERGE INTO section, source of this lesson's reference SQL block. iceberg.apache.org/docs/latest/spark-writes/#merge-into. In English.
  • GitHub — apache/iceberg#15238, the real, documented incompatibility explaining why this lesson's SQL section is representative. github.com/apache/iceberg/issues/15238. In English.
  • This same guide, module 6, lesson 7 — source of the complete five-factor criterion for choosing between SQL MERGE INTO and Python table.upsert(). ../module-06-merge-into-and-native-upserts/en/07-choosing-between-sql-merge-and-python-upsert.md. In English.
  • This guide's DESIGN doc — the full map of the eight modules, including the lesson 7 that follows. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.
← PreviousEvolving and partitioning at scaleNext →What Kiosko still needs