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

Event-time windowed counts with a watermark into Iceberg

Produce a streaming fact: the number of events per 5-minute event-time window, per company. Append the result to an Iceberg table when each window finalizes. A watermark bounds state and defines how late is too late. Spark drops events that arrive after their window has finalized.

Implement windowed_counts(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 10 minutes on the event-time column event_ts (withWatermark("event_ts", "10 minutes")). Do not use processing time.
  • Aggregate into 5-minute event-time windows per company_id, and count the events.
  • Write the result to the existing Iceberg table with the columns company_id, w_start (the window start timestamp), and n (the count). Use append output mode. It is legal because the watermark finalizes each window exactly once.
  • Use checkpoint as the checkpointLocation and trigger(availableNow=True). Call processAllAvailable() and then stop().

The test runs this several times with the same checkpoint. It does these steps:

  1. It writes the events of an early window.
  2. It writes a far-future event. This event moves the watermark past the early window and finalizes it.
  3. It writes a late event for the window that is already finalized.

The final table must contain the count of each finalized window exactly once. The late event must not be counted. The table must match a DuckDB oracle that the test computes over only the in-window (non-late) events.

Starter code

solution.py
"""SPK-11 / 01-watermark-window — starter. See README.md.

Build an event-time windowed count with a 10-minute watermark and append finalized
windows (company_id, w_start, n) to the Iceberg table. Reuse the checkpoint.
"""

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


def windowed_counts(spark, source_dir, table, checkpoint):
    # TODO: readStream JSON (SCHEMA); withWatermark("event_ts","10 minutes");
    # groupBy(window("event_ts","5 minutes"), "company_id").count();
    # select company_id, window.start AS w_start, count AS n;
    # writeStream append -> toTable(table), 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 a watermark on the EVENT-TIME column: .withWatermark('event_ts', '10 minutes'). Never use current_timestamp()/processing time.

Hint 2

Aggregate: groupBy(window('event_ts','5 minutes'), 'company_id').count(). Project to the table columns: company_id, w_start = window.start, n = count.

Hint 3

Append the finalized windows to the Iceberg table with .writeStream.outputMode('append').option('checkpointLocation', checkpoint).trigger(availableNow=True).toTable(table); processAllAvailable(); stop(). Reuse the same checkpoint so late events (arriving after a window finalizes) are correctly dropped.

Run it locally
uv run de start SPK-11/01-watermark-window
# edit workspace/SPK-11/01-watermark-window/
uv run de test