How do I handle late data with watermarks?

Summary

• Call withWatermark(eventTime, delayThreshold) on the streaming DataFrame, before the aggregation and on the same event-time column the aggregation uses. At the start of each trigger the engine sets the watermark to the maximum event time it has seen minus the delay threshold.
• Data within the threshold is aggregated. Data beyond it may be dropped, but Spark guarantees neither outcome, so treat the threshold as tolerated lateness, not a hard cutoff.
• The watermark tells the engine when a window is final, which Append mode needs, and lets it drop old state. Complete mode cannot use it to drop state.
• A longer delay tolerates more lateness but delays Append output and keeps state longer.
• Check numRowsDroppedByWatermark in the query progress to see whether late data is being dropped.

How do I handle late data with watermarks?

A watermark tells Structured Streaming how long to wait for late events before it treats an event-time window as final. The PySpark docs describe it as "a point in time before which we assume no more late data is going to arrive." Spark uses it to know when a window aggregation can be finalized and to minimize the state it keeps for ongoing aggregations. Without one, old aggregation state is not dropped, and Append mode is not supported for aggregations.

How does the watermark move?

Spark computes the watermark from the maximum event time seen across all partitions minus your delay threshold, and sets it at the beginning of each trigger. In the Spark docs' example with a 10-minute threshold, once the engine sees an event at 12:14, the watermark for the next trigger is 12:04. A window ending at time T keeps accepting late data until the maximum event time seen minus the threshold passes T.

How do I add a watermark to a windowed aggregation?

from pyspark.sql.functions import col, window

# events: a streaming DataFrame with an event_time timestamp column and a url column
page_views = (
    events
    .withWatermark("event_time", "10 minutes")
    .groupBy(window(col("event_time"), "5 minutes"), col("url"))
    .count()
)

query = (
    page_views.writeStream
    .outputMode("append")
    .format("console")
    .option("checkpointLocation", "/tmp/checkpoints/page_views")
    .start()
)

The placement rules from the Spark docs:

• Call withWatermark on the same column the aggregation uses, and call it before the aggregation.
• Use Append or Update output mode. Complete mode preserves all aggregate data, so it cannot use the watermark to drop state.
• On a non-streaming Dataset, withWatermark is a no-op.
• The time column must be a timestamp. window() requires TimestampType or TimestampNTZType.

Which output mode should I use?

Output mode	What is written to the sink	Role of the watermark
Append (default)	Only final counts, once the watermark passes the window's end	Drops old state; output is delayed by the threshold
Update	Only rows updated since the last trigger	Drops old state
Complete	The whole Result Table at every trigger	None; state is never dropped

Choose Append for sinks that cannot take fine-grained updates, such as files. Choose Update when you want results early and revisable until the window falls behind the watermark.

What happens to data that arrives after the watermark?

The guarantee is strict in one direction only. Data within the threshold is never dropped. Data delayed beyond it "is not guaranteed to be dropped; it may or may not get aggregated." The PySpark API docs add: "In some cases we may still process records that arrive more than delayThreshold late." Do not write logic that depends on a late row being dropped or kept.
Choose the delay as a latency and correctness trade-off. The Databricks docs say shorter watermarks lower query latency because queries store less state but tolerate little late data, while longer watermarks tolerate late data well but increase query latency.

How do I tell whether late data is being dropped?

Each entry in stateOperators of the query progress reports numRowsDroppedByWatermark, and eventTime.watermark shows the watermark used in the trigger.

progress = query.lastProgress  # a dict, or None before the first progress update
if progress:
    for op in progress["stateOperators"]:
        print(op["operatorName"], op["numRowsDroppedByWatermark"])

The Databricks docs define the metric as "The number of rows that are considered too late to be included in a stateful aggregation." For streaming aggregations it counts rows after aggregation, not raw input rows, and "is not precise, but provides an indication that there is late data being dropped."

What changes with several streams, joins or deduplication?

• Several input streams: Spark tracks the maximum event time per stream and by default uses the minimum of the resulting watermarks as the global watermark. Setting spark.sql.streaming.multipleWatermarkPolicy to max (default is min) lets the watermark move at the pace of the fastest stream, but data from the slower streams is aggressively dropped.
• Outer joins: the watermark and event-time constraints "must be specified", so the engine knows when a row will not match in the future.
• Deduplication: dropDuplicatesWithinWatermark requires a watermark. The docs encourage a delay threshold longer than the maximum timestamp difference among duplicated events.

How does Databricks handle watermarks?

The semantics match open-source Spark. The Databricks docs say: "Queries always process records that arrive within the threshold. Queries might still process records that arrive outside the threshold, but this isn't guaranteed." They also advise referencing columns by name in groupBy() and window() "to ensure the event time marker is preserved." In Lakeflow Pipelines you can write withWatermark("timestamp", "3 minutes") in Python or WATERMARK timestamp DELAY OF INTERVAL 3 MINUTES in SQL, and the docs state: "To ensure queries that perform aggregations are processed incrementally and not fully recomputed with each update, you must use watermarks."

FAQs

Why does my Append-mode query write nothing for a while?

In Append mode a window is written only after the watermark passes its end, so output is delayed by the threshold. In the Spark docs' example, the final counts for the 12:00 to 12:10 window are appended only after the watermark reaches 12:11. The watermark also stalls when no newer events arrive.