What is a watermark in Structured Streaming?

Summary

A watermark in Apache Spark Structured Streaming is a point in event time before which the engine assumes no more late data will arrive. Spark computes it as the maximum event time seen so far minus a delay threshold that you set with withWatermark(). The engine uses it to finalize windows and to drop old state in aggregations, joins and deduplication, so long-running queries keep bounded state.

What is a watermark in Structured Streaming?

A watermark is a moving point in event time before which Spark assumes no more late data will arrive. You declare it with withWatermark(eventTime, delayThreshold): the column that holds the event time, and how long to wait for late records, such as "10 minutes". Event time is the time embedded in the data itself, not the time Spark receives the record. At the start of each trigger, the watermark is the maximum event time seen so far minus the delay threshold. Watermarking has been available since Spark 2.1.
It exists to bound state. A long-running windowed aggregation must keep partial results for old windows, because late data could still update them. For a window ending at time T, the engine keeps that state until the maximum event time seen minus the late threshold is greater than T, then drops it.

How does withWatermark() work?

Concept	What it is	Why it matters
Delay threshold	The second argument to withWatermark()	Sets how long the engine waits for late records
Watermark	Maximum event time seen minus the delay threshold	Windows that end before it can be finalized
State	Partial aggregates, buffered join input or past records kept for deduplication	Without a watermark, old entries are not dropped

The guarantee runs in one direction. Data less than the delay threshold behind the latest data processed is guaranteed to be aggregated. Older data is not guaranteed to be dropped: it may or may not get aggregated, and the more delayed it is, the less likely the engine is to process it.

What does a watermark example look like?

This query counts words in 10 minute windows that slide every 5 minutes, with a 10 minute delay threshold. Columns are referenced by name, as the Databricks docs advise, so the event-time marker is preserved.

from pyspark.sql.functions import window

# words: streaming DataFrame of schema { timestamp: Timestamp, word: String }
windowedCounts = (
    words.withWatermark("timestamp", "10 minutes")
    .groupBy(window("timestamp", "10 minutes", "5 minutes"), "word")
    .count()
)

query = (
    windowedCounts.writeStream
    .outputMode("append")
    .format("console")
    .start()
)

The Spark guide traces a query like this one. When the engine sees (12:14, dog), the watermark for the next trigger is 12:04. A late (12:09, cat) is still ahead of 12:04, so it updates the 12:00 - 12:10 and 12:05 - 12:15 windows. When the watermark reaches 12:11, the 12:00 - 12:10 state is cleared, later records for it are treated as too late, and Append mode writes its final count.

How do output modes interact with watermarks?

Output mode	What the sink receives	Old aggregation state
Append (default)	Final window results only, written after the watermark passes the window, so output is delayed by the threshold	Dropped after the threshold passes
Update	Rows updated since the last trigger; a window can be written again until it is finalized	Dropped after the threshold passes
Complete	The whole result table after every trigger	Never dropped

Complete mode preserves all aggregate data, so a watermark cannot drop state in it. File sinks support only Append.

Which operations use watermarks?

The Spark guide describes these uses:

• Aggregations: group on the event-time column or on a window over it.
• Stream-stream joins: the watermark is optional for inner joins, and required with event-time constraints for outer and semi joins.
• Deduplication: dropDuplicates with the event-time column bounds state, and dropDuplicatesWithinWatermark deduplicates within the watermark.

With several watermarked inputs, the engine uses one global watermark: the minimum by default, or the maximum if spark.sql.streaming.multipleWatermarkPolicy is max, which drops data from the slower streams.

What are the common watermark mistakes?

Symptom	Likely cause	Fix
State is never dropped	Complete output mode	Use Append or Update
Query is invalid in Append mode	The watermark column differs from the aggregation column, or withWatermark() comes after the aggregation	Call it on the same column, before the aggregation
Valid records dropped as late after starting from a Delta table	The initial snapshot is read by file modification time	Set the withEventTimeOrder option

How Databricks approaches watermarks

A watermark is an Apache Spark feature, and the Databricks docs describe the same one-directional guarantee. In Lakeflow Pipelines you define the watermark yourself, with withWatermark() in Python or the WATERMARK clause in SQL, such as WATERMARK timestamp DELAY OF INTERVAL 3 MINUTES. Aggregations in pipelines must use watermarks to be processed incrementally rather than fully recomputed with each update. For large state, Databricks recommends RocksDB-based state management, and serverless pipelines manage state store configuration automatically.
To watch a watermark at work, read eventTime.watermark and stateOperators.numRowsDroppedByWatermark from query.lastProgress. The dropped-rows count is not precise, and for aggregations it counts rows dropped after aggregation, not raw input rows.

FAQs

How do I choose the delay threshold?

A short threshold writes results sooner and keeps less state but tolerates less late data. A long threshold tolerates more late data but holds more state and waits longer to write results. For dropDuplicatesWithinWatermark, set it greater than the maximum timestamp difference between duplicate events.

Does a watermark give exactly-once results?

No. A watermark controls late data and state cleanup, not delivery guarantees. Exactly-once processing comes from replayable sources, idempotent sinks, and checkpointing with write-ahead logs, and Structured Streaming does not deduplicate records from sources. The Spark docs call continuous processing experimental with at-least-once guarantees, and list only map-like operations as supported, not aggregations.