What is a Watermark?

How it works

A streaming job sees events in processing time (when they arrive) but usually aggregates by event time (when they happened). Events arrive late and out of order, so the engine needs a rule for when a window such as “10:00 to 10:10” is finished. Without one it would keep every window’s state forever and could never emit a final result in append mode.

The common rule is:

watermark = max(event_time seen so far) - allowed lateness

In Spark Structured Streaming you declare it with .withWatermark("event_time", "10 minutes") before a windowed aggregation. When the watermark moves past a window’s end, Spark writes that window once (in append mode) and drops its state. An event older than the watermark whose window is already closed is dropped and never corrects the result. Spark keeps the watermark in the checkpoint, so it survives restarts and scheduled availableNow runs.

The allowed lateness is the main tuning decision. A larger value counts more late events correctly but emits results later and holds more state in memory. Pick it from how late real events arrive, and send anything later to a side output or a batch correction.

In batch pipelines the word “watermark” also describes the high watermark of an incremental load: the largest updated_at already loaded, stored so the next run knows where to start. Same idea, different job.

Example

A small simulation of the rule with 10 minutes of allowed lateness and 10-minute windows. The event at 12:03 arrives after the watermark has passed 12:10, so it is too late for its window.

from datetime import datetime, timedelta

LATENESS = timedelta(minutes=10)
arrivals = ["12:01", "12:04", "12:12", "12:08", "12:25", "12:03"]

watermark = None
max_seen = None
for t in arrivals:
    ts = datetime.strptime(t, "%H:%M")
    window_end = ts.replace(minute=ts.minute // 10 * 10) + timedelta(minutes=10)
    if watermark is not None and window_end <= watermark:
        print(f"{t}: window closed at watermark {watermark:%H:%M}, dropped")
        continue
    max_seen = ts if max_seen is None else max(max_seen, ts)
    watermark = max_seen - LATENESS
    print(f"{t}: accepted, watermark now {watermark:%H:%M}")

12:01: accepted, watermark now 11:51
12:04: accepted, watermark now 11:54
12:12: accepted, watermark now 12:02
12:08: accepted, watermark now 12:02
12:25: accepted, watermark now 12:15
12:03: window closed at watermark 12:15, dropped

Common mistakes

• A windowed aggregation with no watermark, whose state grows until the job runs out of memory.
• Expecting append mode to emit a window as soon as it has data; it waits until the watermark passes the window’s end.
• Setting lateness from a guess instead of measuring how late events really arrive.
• Assuming a dropped late event will fix the published total later; once a window is final, it is not revised.