DE Mastery GitHub

Apache Spark · SPK-11

Structured Streaming II — watermarks, state, and streaming SCD2

From your world. SCD2 dimension snapshots and late-arriving facts

← All exercises

Streaming SCD2 upsert into Iceberg with foreachBatch + MERGE INTO

You must fold a change stream of customer-tier updates into a SCD2 dimension in Iceberg, in streaming. Each change does one of three things:

  • It starts a new customer.
  • It changes an existing customer (close the old version, open a new one).
  • It repeats the current value (do nothing).

This is your Kimball SCD2, streamed. foreachBatch gives you a normal batch DataFrame for each micro-batch, so you can run Iceberg MERGE INTO.

Implement upsert_scd2(spark, source_dir, table, checkpoint) in solution.py. It must do these things:

  • Read source_dir as a streaming JSON source with schema cust_key INT, attr STRING, change_ts TIMESTAMP.
  • Use foreachBatch to apply each micro-batch to the existing Iceberg table. Its schema is cust_key INT, attr STRING, valid_from TIMESTAMP, valid_to TIMESTAMP, is_current BOOLEAN.
  • Within the batch, use one change per cust_key (deduplicate the batch on cust_key). Then:
    • Close the current open row when the attribute changed. Set its valid_to to the change_ts of the change and set is_current = false.
    • Insert a new open row for a brand-new key or a changed key: valid_from = change_ts, valid_to = NULL, is_current = true.
    • An unchanged repeat (same attr as the current open row) must change nothing.
  • Use checkpoint as the checkpointLocation and trigger(availableNow=True). Then call processAllAvailable() and stop().

The test runs several times (new key, change, no-op) and asserts the SCD2 invariants:

  • Exactly one open row (valid_to IS NULL and is_current = true) per cust_key.
  • History is preserved. Superseded versions remain with a non-null valid_to and is_current = false (versioning, not overwriting).
  • The current attribute per key equals the latest change. This matches a DuckDB oracle.
  • The valid_to of a superseded row equals the valid_from of the next version (no gaps/overlaps).

Tip: a clean way to do this is two MERGE passes. One pass closes the changed rows. The other pass inserts the new open rows.

Starter code

solution.py
"""SPK-11 / 03-streaming-scd2-merge — starter. See README.md.

Fold the change stream into a SCD2 Iceberg dimension using foreachBatch + MERGE INTO.
Close the old version and open a new one on a real change; do nothing on a repeat.
"""

SCHEMA = "cust_key INT, attr STRING, change_ts TIMESTAMP"


def upsert_scd2(spark, source_dir, table, checkpoint):
    # TODO: readStream JSON (SCHEMA). In foreachBatch(fn):
    #   - dedup the batch on cust_key, register it as a temp view,
    #   - MERGE pass 1: close current open rows whose attr changed
    #     (set valid_to = change_ts, is_current = false),
    #   - MERGE pass 2: insert a new open row for keys not currently open
    #     (valid_from = change_ts, valid_to = NULL, is_current = true).
    # Write with checkpointLocation=checkpoint, trigger(availableNow=True);
    # processAllAvailable(); stop().
    raise NotImplementedError

Hints

Hint 1

Read the JSON change stream (schema 'cust_key INT, attr STRING, change_ts TIMESTAMP') and write it with .foreachBatch(fn). Inside fn, make a batch DataFrame view of the micro-batch (one change per cust_key: dropDuplicates(['cust_key'])) and run Iceberg MERGE INTO against the dimension table.

Hint 2

Do it in two MERGE passes. Pass 1 CLOSES the current open row when the attr changed: MATCHED on t.cust_key=c.cust_key AND t.is_current AND t.attr<>c.attr -> UPDATE SET valid_to=c.change_ts, is_current=false. Pass 2 INSERTS a new open row: NOT MATCHED on (cust_key, is_current) -> INSERT valid_from=c.change_ts, valid_to=NULL, is_current=true.

Hint 3

Reuse the checkpoint and trigger(availableNow=True); processAllAvailable(); stop(). Invariants: exactly one open (valid_to IS NULL / is_current) row per cust_key, old versions kept with a closed valid_to (versioning, not overwrite), unchanged repeats change nothing.

Run it locally
uv run de start SPK-11/03-streaming-scd2-merge
# edit workspace/SPK-11/03-streaming-scd2-merge/
uv run de test