What is watermarking in Structured Streaming?

A watermark tells Structured Streaming how late event data may arrive and still be counted. Spark tracks the latest event time it has seen, subtracts the delay you set, and treats that point as the watermark: windows and state older than it are finalised and dropped, and records older than it may be ignored. It is what keeps state from growing forever and what lets append mode emit final results.

from pyspark.sql import functions as F

clicks = spark.readStream.table("bronze.clicks")    # has event_time

per_window = (clicks
    .withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "5 minutes"), "page")
    .count())

(per_window.writeStream
    .outputMode("append")                             # emit each window once, when final
    .option("checkpointLocation", "/chk/clicks_5min")
    .toTable("silver.clicks_5min"))

How it works

1. In each micro-batch, Spark finds the maximum event_time seen so far, say 12:30.
2. The watermark is that minus the delay: 12:20. It is updated between micro-batches.
3. A 5-minute window ending at or before 12:20 can no longer receive data, so Spark writes its final count (in append mode) and removes its state.
4. A record with event time 12:14 arriving now belongs to a closed window and is dropped.

The guarantee, precisely Data that is less late than the delay is guaranteed to be included. Data later than that may or may not be included: it is not guaranteed to be dropped straight away, only eventually.

Choosing the delay It's a trade-off between completeness and latency and state size. Base it on how late your data really arrives, for example the 99th percentile of the difference between event time and arrival time. Too short loses late events; too long holds state and delays final results.

Where watermarks are needed or useful

• Windowed aggregations in append mode, which can only emit a window once it is final.
• Stream–stream joins, to bound how long each side's rows are kept; outer joins require them.
• Deduplication: dropDuplicates with a watermark, or dropDuplicatesWithinWatermark (Spark 3.5 and later), keeps only recent keys in state.
• Custom stateful processing with transformWithState, which can use event-time timers.

Conditions

• The watermark column must be the event-time column used in the aggregation, and withWatermark must be called before the aggregation.
• Output mode must be append or update; complete mode keeps all state, so watermarks don't clean anything up.
• With several input streams, Spark by default uses the minimum of their watermarks (spark.sql.streaming.multipleWatermarkPolicy set to min), so the slowest stream decides.

Monitoring The query's progress (query.lastProgress) reports the current watermark and the number of rows dropped as too late, which tells you whether the delay is right.