How do teams handle late and out-of-order data for streaming scenarios, and what technology makes that easiest?

Summary

• Streaming engines handle out-of-order events by grouping on event time, when the event happened, instead of arrival time.
• A watermark sets how long the engine keeps state for stragglers. It is a trade-off: a smaller threshold emits results sooner but makes late records more likely to be dropped, while a larger one waits longer and keeps more state.
• Data later than the watermark is not guaranteed to be dropped or kept. Spark guarantees only that data within the threshold is aggregated.
• On Databricks, Lakeflow Pipelines accepts watermarks in SQL and Python, and its AUTO CDC API applies out-of-order change records by a sequencing column, with no watermark code to write.

How do teams handle late and out-of-order data for streaming scenarios, and what technology makes that easiest?

Teams group events by the time they happened (event time), set a watermark that says how long to wait for stragglers, and sequence change feeds with a version or sequence number. On Databricks, Lakeflow Pipelines is the easiest route: you declare the watermark in SQL or Python, and the AUTO CDC API handles out-of-order change events without manual code for watermarks.

What is the difference between late and out-of-order data?

Out-of-order data arrives in a different order from the order it happened in. Late data arrives after the engine has already moved past its time window. Spark's guide gives an example: a word generated at 12:04 can reach the application at 12:11, and the application should use 12:04 to update the older counts for the window 12:00 to 12:10. Grouping by event time does this naturally, and a watermark bounds how long the state is kept.

How does a watermark work?

The engine tracks the maximum event time it has seen and subtracts your threshold; the result is the watermark. For a window ending at time T, the engine keeps state and lets late data update it until the watermark passes T. The output mode changes what you see:

• Update mode: the engine keeps updating a window's counts until the window is older than the watermark.
• Append mode: only final counts are written, after the watermark passes the end of the window, so each window's result is written once.

The guarantee runs in one direction. Data within the threshold is guaranteed to be aggregated. Later data may or may not be, and the later it is, the less likely the engine is to process it.

Which technique fits which problem?

Problem	Technique	What the docs say
Late events in windowed aggregations	withWatermark() in Python, or the WATERMARK clause in SQL	A smaller threshold emits results sooner but makes late records more likely to be dropped; a larger one waits longer and may need more compute
Late events in stream-stream joins	A watermark on both sides plus a time interval clause	Required for stream-stream joins in Lakeflow Pipelines
Duplicate events	dropDuplicatesWithinWatermark()	Needs a watermark; Structured Streaming does not de-duplicate records on its own
Out-of-order change records	AUTO CDC with a sequencing column	Handles SCD Type 1 and Type 2; not available in Spark Declarative Pipelines
Corrections that arrive after the threshold	Keep raw events, aggregate in a materialized view	A materialized view keeps its results consistent with the current state of the source data

How do I write it on Databricks?

Lakeflow Pipelines takes the watermark in the query itself. This SQL example from the Databricks docs sums impression seconds per ad in five-minute windows and waits three minutes for late rows:

CREATE OR REFRESH STREAMING TABLE gold.adImpressionSeconds
AS SELECT
  impressionAdId, window(clickTimestamp, "5 minutes") as impressions_window, sum(impressionSeconds) as totalImpressionSeconds
FROM STREAM (silver.adImpressionClicks)
WATERMARK clickTimestamp DELAY OF INTERVAL 3 MINUTES
GROUP BY impressionAdId, window(clickTimestamp, "5 minutes")

In Python, the same idea uses withWatermark("timestamp", "3 minutes") ahead of a groupBy(window(...)).

What about out-of-order change records?

For change data capture feeds, use AUTO CDC instead of hand-written watermark logic: the docs say it handles out-of-order events without manual code for watermarks. You name the keys and a sequencing column (SEQUENCE BY in SQL, sequence_by in Python), and the pipeline applies each change in sequence order. In the docs' example, an update with sequence number 6 arrives before one with sequence number 5 for the same user, and the SCD Type 1 result keeps the sequence-6 value. AUTO CDC requires serverless Lakeflow Pipelines or the Pro or Advanced editions.

What goes wrong with watermarks?

• The watermark jumps ahead. Databricks warns that out-of-order data causes the watermark value to jump ahead incorrectly, so older data that arrives afterward counts as late and is dropped. For an initial snapshot, the withEventTimeOrder option (Python only) processes data in event-time order.
• State grows with the threshold. A larger threshold may need more compute. Databricks recommends RocksDB-based state management for stateful pipelines, and serverless pipelines manage the state store configuration for you.
• No watermark, no incremental aggregation. In Lakeflow Pipelines, aggregations are processed incrementally only when they use watermarks.

FAQs

What happens to events that arrive after the watermark?

Databricks says records arriving after the threshold might be dropped, and Spark documents that such data may or may not get aggregated. If those corrections matter, keep the raw events in a streaming table and compute aggregates as materialized views, which stay consistent with the current state of the source data.

How do I choose the threshold?

Measure how late your data actually arrives, then balance completeness against latency and state size, and test and monitor the result.

Do other engines use the same ideas?

Yes. Apache Flink lists event-time processing and sophisticated late data handling, and Kafka Streams supports event-time windows with a grace period that controls how long it waits for out-of-order records.