Structured Streaming II — watermarks, state, and streaming SCD2

The watermark trails the maximum observed event time by a fixed delay. When it passes the end of a window, the window finalizes (Spark can append its result) and Spark frees its state. Spark treats data older than the watermark as late and drops it. This is a trade-off between completeness and latency. It is not a guarantee of no loss.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#handling-late-data-and-watermarking

Event time is a property of the data. The same events give the same windows, no matter when or how Spark processes them. Processing-time windows change with scheduling. This breaks reproducibility (and auditability).

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#window-operations-on-event-time

Without a watermark, the groups of an aggregation never finalize, so Spark rejects append. With a watermark, a window becomes final when event time passes its end + delay. Spark appends that final row once.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#output-modes

Plain dropDuplicates keeps all seen keys without a time limit (unbounded state). The watermarked form forgets keys older than the horizon. This bounds the state. The cost is that it does not recognize a duplicate that arrives after the horizon. This is consistent with the watermark contract.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#streaming-deduplication

SCD2 keeps exactly one current/open version per natural key. It closes earlier versions with a non-null valid_to. If you leave several rows open (you forget to close the old one), you have a classic SCD2 bug.

Source: https://iceberg.apache.org/docs/latest/spark-writes/#merge-into

foreachBatch gives you the micro-batch as a batch DataFrame and a batch_id, so any batch operation (including Iceberg MERGE INTO) works. The streaming sinks cannot express a merge.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#foreachbatch

Overwriting in place is SCD type 1: no versioning, no history. SCD2 must close the old version and insert a new one, so that you can still query the earlier states.

Source: https://iceberg.apache.org/docs/latest/spark-writes/#merge-into

The watermark trails the maximum event time by 10 minutes. So it passes the window end (00:05) only when Spark sees an event with event_ts >= 00:15. After that, Spark emits the window, and later events for it are late.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#handling-late-data-and-watermarking