Why is late data being dropped unexpectedly?

Summary

• Without a watermark, a Structured Streaming aggregation doesn't drop late rows. It keeps state for old windows so late rows can still update them, and that state keeps growing. Event-time drops start only after you declare a watermark with withWatermark.
• The watermark is the maximum event time seen minus your delay threshold. It follows the data, not the clock.
• The guarantee runs one way. Rows delayed by less than the threshold are aggregated. Rows delayed by more may or may not be.
• Unexpected drops usually mean the threshold is shorter than the real lateness, or something pushed the maximum event time forward. The query progress reports the count as numRowsDroppedByWatermark.
• In Append mode a windowed result is emitted only after the watermark passes the window, so a longer threshold also delays output.

Why is late data being dropped unexpectedly?

A watermark told Spark to stop waiting. Without one, Structured Streaming "can maintain the intermediate state for partial aggregates for a long period of time such that late data can update aggregates of old windows correctly", according to the Spark docs. Once you call withWatermark, the engine tracks the maximum event time it has seen, subtracts your delay threshold, and cleans up state older than that point. An unexpected drop therefore usually means the threshold is shorter than your real lateness, or the maximum event time jumped ahead of where you expected.

How does a watermark decide what is late?

Event time is "the time embedded in the data itself", not the time Spark receives the row. The watermark is set to the maximum event time minus the threshold at the beginning of every trigger. For a window ending at time T, the engine keeps state and lets late data update it until the watermark passes T.
The Spark docs walk through a 10-minute threshold. The engine sees the row (12:14, dog), so the watermark for the next trigger is 12:04. The out-of-order row (12:09, cat) is still ahead of 12:04, so it updates its windows. Once the watermark reaches 12:11, the state for the window from 12:00 to 12:10 is cleared, and a later row such as (12:04, donkey) is considered "too late" and ignored.
The guarantee holds in one direction only. Data delayed by less than the threshold is aggregated. Data delayed by more "may or may not get aggregated", and the later it arrives, the less likely the engine is to process it.

Why does late data disappear?

Symptom	Likely cause	Fix
Rows missing from windows that already closed	The threshold is shorter than the real lateness	Raise the threshold, then confirm with numRowsDroppedByWatermark
Normal rows are dropped right after one odd record	The watermark follows the maximum event time across partitions, so a far-future timestamp pushes it forward	Filter or correct implausible event times before withWatermark
A slow input loses rows in a join	spark.sql.streaming.multipleWatermarkPolicy is set to max, so the global watermark follows the fastest input and drops data from slower streams	Use the default, min, which moves at the pace of the slowest stream
The watermark seems to do nothing	withWatermark was called on a different column than the aggregation, after the aggregation, or on a batch DataFrame, where it is a no-op	Call it on the same event-time column of the streaming DataFrame, before groupBy
Late files are never read by a file source	The maxFileAge option ignores files older than its default of 1 week, measured against the latest file's timestamp	Raise maxFileAge to cover how old a late file can be

How do you find out what is being dropped?

from pyspark.sql import functions as F

events = ...  # streaming DataFrame with event_time (timestamp) and user_id

counts = (events
    .withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "1 hour"), "user_id")
    .count())

query = counts.writeStream.outputMode("update").format("console").start()

progress = query.lastProgress
if progress:
    print(progress["eventTime"])  # keys: max, min, avg, watermark
    for op in progress["stateOperators"]:
        print(op["operatorName"], op["numRowsDroppedByWatermark"])

The Spark source describes numRowsDroppedByWatermark as "Number of input rows dropped because their event time was older than the watermark." Each stateful operator reports its own count.

What changes for deduplication and joins?

Deduplication uses the watermark too. Deduplicate on the ID and the event-time column so the query can drop old state, or use dropDuplicatesWithinWatermark(["guid"]), which requires a watermark, when duplicates carry different event times. In stream-stream joins the watermark and an event-time constraint are optional for inner joins and mandatory for outer joins, whose NULL results arrive with a delay.

How does Databricks approach late data?

Databricks documents the same guarantee: "Queries always process records that arrive within the threshold. Queries might still process records that arrive outside the threshold, but this isn't guaranteed." Lakeflow Pipelines use watermarks for aggregations, joins and deduplication, declared as withWatermark("timestamp", "3 minutes") in Python or WATERMARK timestamp DELAY OF INTERVAL 3 MINUTES in SQL. Auto Loader "does not guarantee the order in which files are discovered or processed", so expect out-of-order event times when you ingest files.

FAQs

Does a watermark add latency?

In Append mode, yes. The Spark docs say the output of a windowed aggregation "is delayed the late threshold specified in withWatermark()", because rows are added to the result only once they are finalized. In Update mode, updated rows are written after every trigger.

How do I pick the threshold?

Start from how late your data really is. Databricks docs say "Choosing a smaller threshold results in records being emitted sooner but also means late records are more likely to be dropped."

Can I recover rows that were already dropped?

Not from the stream's state. To recompute totals that include very late rows, run a batch query over the stored source data. The Spark docs say a watermark has no effect on batch queries.