Apache Spark / Advanced: streaming and recoverable operation / Lesson 13

Streaming event time and watermarks

Build incremental time-window results while making late data, state retention and output-mode assumptions explicit.

About 4 min readContent reviewed 2026-10-28

On this page

Lesson goals

Distinguish event time, processing time and trigger cadence.
Explain watermark progress without promising a universal lateness deadline.

An unbounded input needs a result-update model

Structured Streaming represents an unbounded input through DataFrame-style operations that Spark processes incrementally. The default micro-batch model is different from repeatedly running an unrelated batch query over every file. Each increment must relate to source progress and retained state. A reading’s event_time records when the meter observed production; processing time records when the system handles it. Reconnection can make these diverge substantially. Choose the business clock before grouping windows, and define how a published total changes when an old reading arrives. A streaming API does not remove the need for source identity and completeness rules.

Watermark the same event-time column used by the stateful rule

The fragment assumes an existing streaming events relation with a timestamp event_time. It requests ten minutes of watermark delay and five-minute site windows. The placement and supported operation determine whether Spark can use the watermark for state cleanup. These durations are a synthetic policy, not a recommendation for all meters. A window sum over accepted inputs also needs a prior rejection policy for malformed or implausible times. A sensor accidentally reporting far-future time can distort progress, so source validation and operational monitoring belong beside the window definition.

python Illustrative technical example; logical results are reasoned, not engine-tested.

from pyspark.sql import functions as F
windowed = (events.withWatermark('event_time', '10 minutes')
    .groupBy('site_id', F.window('event_time', '5 minutes'))
    .agg(F.sum('kwh').alias('total_kwh')))

A streaming transformation plan only. It starts no query and does not establish a sink or exactly-once publication.

A watermark follows observed event progress

A watermark is derived from observed event-time progress and a configured delay, with engine update and operator rules. The local timeline subtracts ten minutes from a maximum observed time of12:20 to illustrate a candidate boundary12:10. It is not a captured watermark value and must not be treated as an exact instant advancement guarantee. Spark documents a one-sided lateness guarantee for eligible operations: sufficiently recent data is not dropped, while older data may or may not be processed. A watermark is neither a promise to wait exactly ten wall-clock minutes nor an automatic audit ledger of all discarded readings.

text Illustrative technical example; logical results are reasoned, not engine-tested.

maximum observed event time:12:20
configured delay:10 minutes
local candidate calculation:12:10
actual engine watermark:inspect authorized query progress

A local time-subtraction model. Real watermark advancement and eligibility depend on the streaming engine and operator.

Output mode controls what changes reach a sink

Append, Update and Complete modes expose different result changes and have operation-specific support. A window aggregate in append mode can wait for the watermark to establish finality before emitting a row, whereas update-style output can revise aggregates. A consumer must know whether a row replaces a previous total or adds a new contribution. Blindly adding every updated total double-counts energy. Complete mode can retain a broad result state and has different cleanup implications. Validate the chosen mode with late events, empty batches and a stalled source; a short trigger interval alone does not guarantee low end-to-end freshness.

State growth is a correctness and capacity signal

Stateful aggregation, deduplication and stream joins each have their own retention and support constraints. An idle or skewed source can affect watermark progress, and multiple inputs introduce a documented global-watermark policy. Monitor state rows, memory, batch duration and event-time progress together. Do not delete checkpoints or shorten delay merely to make state smaller without understanding the resulting data policy. This course launches no stream, synthetic rate source, socket or state store. It validates only the local timeline arithmetic and clearly distinguishes that check from engine watermark or late-data testing.

Common errors and fixes

A ten-minute watermark is treated as a ten-minute sleep

Why it happens: Event-time progress was confused with wall-clock waiting.
How to fix it: Inspect observed event times and operator watermark semantics.
Prevent it next time: Test idle input and delayed observations in a separate lab.

Updated window totals are added repeatedly

Why it happens: A replacement-style result was consumed as an additive event.
How to fix it: Define the sink key and update semantics for each window.
Prevent it next time: Reconcile corrections and repeated batches explicitly.

State grows despite a configured watermark

Why it happens: The chosen operation, output mode or event progress does not permit expected cleanup.
How to fix it: Review watermark placement, support rules and progress metrics together.
Prevent it next time: Budget state and inspect progress before changing retention policy.

Check your understanding

1. What local candidate follows max event time12:20 and a ten-minute delay?

Show worked solution

The arithmetic candidate is12:10. It is not proof that an actual Spark watermark has already reached that value. Engine update timing, input progress and operator rules remain relevant. Inspect authorized query progress and test the required late-data behavior separately.

2. Why can a consumer double-count an Update-mode window result?