Structured Streaming II — watermarks, state, and streaming SCD2

7 h
requires SPK-10

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

Bounded streaming deduplication with a watermark

An at-least-once source delivers each event one or more times. Write a stream that keeps one row per event_id in an Iceberg table. Use a watermark, so that the dedup state stays bounded. The state forgets ids that fell behind the watermark, instead of remembering every id forever.

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

Read source_dir as a streaming JSON source with schema event_id INT, company_id STRING, amount DOUBLE, event_ts TIMESTAMP.
Declare a watermark of 30 minutes on the event-time column event_ts.
Deduplicate on the business id with dropDuplicatesWithinWatermark(["event_id"]). This is the bounded form. Do not use plain dropDuplicates, which keeps every id forever.
Append the deduplicated rows to the existing Iceberg table (event_id INT, company_id STRING, amount DOUBLE, event_ts TIMESTAMP) in append mode.
Use checkpoint as the checkpointLocation and trigger(availableNow=True). Then call processAllAvailable() and stop().

The test runs this several times with the same checkpoint. Duplicate event_ids appear within the watermark horizon (in the same run and across runs). The final table must contain exactly one row per distinct event_id. It must match a DuckDB oracle that keeps one row per id. It must have no duplicates and no loss of distinct ids.

Starter code

solution.py

"""SPK-11 / 02-dedup-within-watermark — starter. See README.md.

Keep one row per event_id using a watermark-bounded dedup, appending to the Iceberg
table. Reuse the checkpoint.
"""

SCHEMA = "event_id INT, company_id STRING, amount DOUBLE, event_ts TIMESTAMP"


def dedup(spark, source_dir, table, checkpoint):
    # TODO: readStream JSON (SCHEMA); withWatermark("event_ts","30 minutes");
    # dropDuplicatesWithinWatermark(["event_id"]); append -> toTable(table) with
    # checkpointLocation=checkpoint, trigger(availableNow=True); processAllAvailable();
    # stop().
    raise NotImplementedError

Hints

Hint 1

Read the JSON stream (schema 'event_id INT, company_id STRING, amount DOUBLE, event_ts TIMESTAMP'). Add an event-time watermark: .withWatermark('event_ts', '30 minutes').

Hint 2

Deduplicate on the business id with the BOUNDED operator: .dropDuplicatesWithinWatermark(['event_id']). This keeps the first occurrence and forgets the key once it falls behind the watermark, so state stays bounded (unlike dropDuplicates which remembers every id forever).

Hint 3

Append the deduplicated rows to the Iceberg table: .writeStream.outputMode('append').option('checkpointLocation', checkpoint).trigger(availableNow=True).toTable(table); processAllAvailable(); stop(). Reuse the same checkpoint across runs so dedup state carries over.

Run it locally

uv run de start SPK-11/02-dedup-within-watermark
# edit workspace/SPK-11/02-dedup-within-watermark/
uv run de test