Structured streaming watermark architecture — event time, late data, trigger, output

Why architecture matters here

Watermark bugs are hard to spot. Data appears to be counted correctly until a late-arriving event 20 minutes past the window boundary is silently dropped. The architecture matters because event-time semantics require deliberate delay + output mode choices.

With the pieces mapped, you can define correctness for your streaming pipeline explicitly.

The architecture: every piece explained

The top strip is the semantics. Source reads events. Event time column annotates each row. Watermark is (max event time seen) − allowed delay; state older than watermark is safe to close. State store holds windowed aggregations.

The middle row is control. Trigger defines the processing-time cadence (continuous, once, or interval). Output mode — append (only closed windows), update (any changed row), complete (whole state) — must match sink. Late data policy decides whether to drop or route past-watermark data to a side output. Checkpoint to durable storage enables restart.

The lower rows are sinks + ops. Sink: Delta, Kafka, or files. Observability tracks batch duration + lag. Ops covers schema evolution, state pruning, and backfill.

End-to-end flow

End-to-end: job aggregates page views per 5-minute window with watermark delay 10 minutes. Trigger interval 1 minute. State grows during the last 10 minutes; watermark closes older windows and outputs final counts in append mode. A late event 8 minutes past its window still gets counted (inside the delay). A late event 12 minutes past is dropped (or routed via late-data sink). Checkpoints allow restart from last processed offsets. Observability shows batch duration steady + lag under 1 minute.

Event time versus processing time

Two clocks run in every streaming pipeline and they never agree. Event time is the timestamp carried inside the record itself: the instant a phone registered a tap, a sensor sampled a temperature, a gateway authorised a card. Processing time is whatever an executor's clock reads when Spark finally deserialises that row inside a micro-batch. The distance between the two is the entire subject of this article.

If the clocks agreed, windowing would be trivial. A five-minute bucket would mean "rows I received in the last five minutes", and the moment the wall clock crossed the boundary you could close the bucket and forget it. But an offline mobile client flushes an hour of buffered taps when it reconnects. A regional collector retries a failed upload. Three Kafka partitions are consumed in parallel and one of them is four minutes behind the others because a consumer rebalanced. A backfill replays yesterday's topic into the same query. Each of these delivers a record whose event time sits well behind the event times of the records around it.

Bucketing by processing time puts those records in the wrong bucket, and worse, makes the answer depend on how fast your cluster happened to be running that afternoon. Bucketing by event time gives a result that is reproducible: the same input produces the same counts whether you replay it in ten seconds or ten hours, on four executors or forty. That reproducibility costs you one thing, and it is the thing processing time never had to answer - when is a bucket finished? You cannot wait forever, because the memory holding open buckets is finite. You cannot close on schedule, because the straggler may still be in flight. A watermark is what turns that judgement call into arithmetic.

What a watermark actually is

Strip away the metaphor and a watermark in Structured Streaming is a single timestamp, recomputed once per micro-batch, defined by one line of arithmetic:

watermark(after batch N) = max(event time observed in batches 0..N) - delayThreshold

Three properties of that formula do all the work downstream.

It is derived from data, not from a clock. Nothing about the wall clock enters the calculation. The watermark only knows that time has passed because it saw a record that says so. A source that goes quiet freezes the watermark, however long the job keeps running - a consequence severe enough that it gets its own section below.

It is monotonic. The maximum is taken over every batch since the query began, so a batch consisting entirely of old records cannot drag the watermark backwards. Once the engine has declared that it has seen 14:52, it will never un-declare it. This is what makes the watermark usable as a safety boundary: state you released on its authority can never become needed again.

It is applied one batch late. The value is computed at the end of a micro-batch and used at the start of the next one. Batch N filters late rows and evicts state using the number derived from batches 0 through N-1, not from its own contents. Two practical consequences follow. The first batch of a brand-new query runs with the watermark at its initial value, so nothing is discarded no matter how ancient it is - convenient if your first batch is a large historical backlog. And a restarted query resumes the watermark it had recorded rather than starting over, which is why a restart does not suddenly re-admit data it had been rejecting before the restart.

Placing withWatermark so it takes effect

The declaration is one method call, and where you put it decides whether it does anything at all.

events = (spark.readStream.format("kafka")
          .option("subscribe", "impressions")
          .load()
          .select(from_json(col("value").cast("string"), schema).alias("e"))
          .select("e.*"))                      # e.event_ts is a TimestampType

counts = (events
          .withWatermark("event_ts", "10 minutes")     # BEFORE the aggregation
          .groupBy(window(col("event_ts"), "5 minutes"), col("campaign_id"))
          .count())

The rules that bite:

The named column must be a timestamp, and it must be the same column the stateful operator uses as its event-time key. Watermarking event_ts and then windowing on ingest_ts gives you a watermark that tracks one column and an aggregation that is bounded by nothing.

The call must sit upstream of the stateful operator in the same lineage. Attaching it to the output of the aggregation is perfectly legal code and does precisely nothing for that aggregation. So is projecting the event-time column away, or renaming it, between the withWatermark and the groupBy - the link is by column, and if the column does not survive to the operator, neither does the bound.

The delay is an interval string such as "10 minutes" or "1 hour", and it must be non-negative. Zero is allowed and means "close the window the instant its end is reached", which in practice drops anything even slightly out of order.

Append mode refuses to plan a streaming aggregation with no watermark and fails at query start with an analysis error. Update and complete mode start cheerfully, which is exactly how unbounded state reaches production - the query that crashes at submission is the lucky one. And on a purely stateless query, withWatermark is close to inert: there is no state to evict and no late-row filter to insert, so rows flow through untouched.

How the watermark evicts window state

A windowed aggregation keys its state on the window boundaries plus your grouping columns, so groupBy(window(ts, "5 minutes"), campaign_id) holds one state row per open window per campaign. The eviction rule is a comparison, not a timer: a window is finalised and removed once watermark >= window end. At that point no row that could still be admitted can belong to it, so keeping it would be keeping a bucket that can never change.

Work the numbers for the query above - five-minute tumbling windows, a ten-minute delay. The window covering 10:00 to 10:05 is released when the watermark reaches 10:05, which requires a maximum observed event time of 10:15. So at steady state the query holds roughly (delay + window length) / window length generations of windows open, here three, plus whatever is mid-flight. Multiply by distinct campaign ids and you have the state row count, near enough for capacity planning.

Sliding windows change that arithmetic sharply. window(ts, "10 minutes", "1 minute") assigns every record to ten overlapping windows, so the same key cardinality and the same delay produce ten times the state rows and ten times the update work per record. Session windows are worse behaved still, because a session's end is not known in advance and a busy key can keep one alive far past any fixed window length.

What the watermark does not control is where that state lives, how it is checkpointed, or which backend stores it - those belong to the state store, covered in Spark Structured Streaming state. The watermark decides only one thing: when a row of state stops being reachable.

The exact rule for dropping a late row

For a stateful operator the rule is: a row is discarded when event time < watermark, where the watermark is the value in force for that batch - the one computed at the end of the previous batch. A row whose event time is at or after the watermark is admitted and folded into its window's state, however long ago that window opened.

Now notice how the two thresholds interlock. Eviction fires when the watermark reaches a window's end. Dropping fires when a row falls below the watermark. Any row Spark admits therefore has an event time at or above the watermark, which puts its window's end strictly above the watermark, which means that window has not been evicted. Admitted late data always finds its state still alive. That is not luck - it is the reason one number is made to do both jobs.

The rule is per-operator and per-batch, which matters when a query has two stateful operators with different windows: the same row can be comfortably on time for a one-hour aggregation and hopelessly late for a one-minute one further down the plan.

"Discarded" means discarded. There is no automatic dead-letter path; the row is counted in the progress report and then it is gone. If losing those records is unacceptable, branch the stream before the stateful operator and route rows older than your own threshold to a separate sink, or handle the reconciliation in foreachBatch where you still have the raw rows in hand. Both approaches mean maintaining your own notion of lateness alongside the engine's, which is annoying, and is still cheaper than discovering six months later that a chronically lagging producer has been contributing nothing to your revenue numbers.

Output modes and why append feels broken

Append is the mode most teams want, because it maps cleanly onto files, Delta and Kafka, and it is the mode that surprises them. In append mode a windowed aggregate emits nothing until the watermark crosses the window end. The lag between an event happening and its window's count becoming visible is therefore roughly window length + watermark delay + one trigger interval. Five-minute windows, a ten-minute watermark and a one-minute trigger put your dashboard up to sixteen minutes behind, and the query is working perfectly. The predictable mistake is to attack that number by shortening the watermark, which does reduce latency and does so by silently throwing away the late records the threshold was protecting.

Update emits the running value each trigger, so you watch a window's count climb and then settle. The sink must be keyed and idempotent, because the same window will be written repeatedly and the last write must win. Complete rewrites the entire result every trigger, and that is precisely why it can never evict anything: it has committed to being able to reproduce every row. A complete-mode aggregation with a withWatermark on it is a query whose state grows forever and whose watermark is decoration.

Stream-stream joins: watermarks on both sides plus a range condition

Joining two streams is where the watermark stops being a tuning knob and becomes load-bearing. Both sides have to be buffered, because a row on the left can match a row on the right that has not arrived yet, and with no bound the engine has to keep every row from both inputs indefinitely.

Bounding it takes two ingredients, and one without the other is useless: a watermark on the event-time column of each input, and a time-range condition in the join predicate that relates the two event-time columns to one another.

imps  = impressions.withWatermark("imp_ts", "10 minutes")
clks  = clicks.withWatermark("clk_ts", "20 minutes")

joined = imps.join(clks, expr(
    "ad_id = click_ad_id AND "
    "clk_ts BETWEEN imp_ts AND imp_ts + interval 1 hour"))

The range condition is what converts a watermark into an expiry date for a buffered row. Because a click can only match an impression at most an hour older, once the click-side watermark reaches T the engine can prove that no future click will match an impression stamped before T minus one hour - so those impressions leave the buffer. Drop the range condition and the watermark has nothing to reason with. The query still plans, still returns correct rows, and still accumulates both inputs until it dies.

For inner joins the pair is technically optional, which is the trap: you get a correct query with a fatal memory profile and no warning. For outer joins it is mandatory and the planner rejects the query without it, because emitting a NULL-padded row is a claim that no match will ever arrive, and only a watermark plus a range bound can justify that claim. Expect the corresponding behaviour at runtime: unmatched outer rows appear late, when the bound expires, not in the trigger where the row showed up.

Multiple watermarks in one query: min versus max

A query with several streaming inputs carries several watermarks, one per input, each advancing at the pace of its own data. The stateful operators need one number. Spark reconciles them into a global watermark and the reconciliation policy is a config, spark.sql.streaming.multipleWatermarkPolicy, which defaults to min.

Minimum is the safe default for a concrete reason: the global watermark never overtakes the slowest input, so no input ever has its records ruled late on the strength of data it never produced. The price is that the slowest input governs everything - state retention, append-mode latency, join buffer lifetime - for every operator in the query. One low-volume, high-lag stream sets the pace for the whole job.

Switching the policy to max lets the fastest input drive. State shrinks, append output arrives sooner, join buffers turn over quickly. It also means the lagging input's records now arrive behind a watermark advanced entirely without them, and they are dropped as late. There are situations where that is the right trade - a trickle of reference updates joined against a firehose, where hours of buffered state costs more than the occasional missed update - but it is a data-loss setting, it applies to the whole query rather than to one operator, and it should be a decision someone wrote down rather than a config someone copied.

Choosing the delay: state, completeness, latency

delayThreshold is effectively the only knob, and it moves three quantities at once in opposite directions. Increase it and you count more late records, hold more windows open, and publish append-mode results later. Decrease it and you get the mirror image. There is no setting that is good at all three; there is only the setting that is right for what your consumers actually need.

The state cost is easy to estimate and worth estimating before you change it. Open generations are about (delay + window length) / slide. On five-minute tumbling windows a ten-minute delay keeps three generations open; moving to twenty minutes keeps five. That is a 67 percent increase in state rows for the same traffic, and if your state store is already the reason your batches are slow, it will show up as slower batches rather than as a memory error.

Pick the number from measurement rather than intuition. Take a representative day of raw records, compute the difference between arrival time and event time for each one, and look at the distribution's tail. The p99 tells you what routine skew looks like; the p99.9 tells you what the stragglers cost. Setting the threshold near p99.9 usually buys a modest amount of extra state and converts "one event in a thousand vanishes without trace" into "it is counted". Sizing for the worst case of a bad day is a different decision - that is sizing continuous state for an event that happens monthly, and a backfill is better handled as a separate batch job against the same sink than by carrying its buffer all year.

When the watermark stops moving

Spark does not maintain a watermark per input partition. It takes the maximum event time across every row in the micro-batch and derives a single number. That design decision has two edges and both of them are sharp.

The helpful edge: one quiet Kafka partition in an otherwise busy topic stalls nothing, because the other partitions keep supplying event times and the maximum keeps climbing. The harmful edge is the same fact seen from the other side. When that quiet partition wakes up, its backlog arrives behind a watermark that the busy partitions advanced without it, and every one of those records is dropped as late. A partition lagging by more than your delay threshold is losing data continuously while the query looks entirely healthy - throughput normal, batches on time, no errors. This is a genuine difference from engines that track watermarks per partition and combine them; Flink takes the other approach, with its own cost, and the contrast is worth understanding if you operate both.

The outright stall arrives when a whole input goes quiet. The maximum stops moving, the watermark freezes at its last value, and every open window stays open. Under the default minimum policy a single dead input pins the global watermark while the other streams keep feeding state that will never be released. Nothing errors, because from the engine's point of view it has simply received no evidence that time has advanced.

Since the watermark will not advance from wall time, the remedies are operational rather than configuration. Keep a low-rate heartbeat record flowing on every source so the maximum always moves. Alert on the gap between now and the reported watermark rather than on error counts. And where a genuinely intermittent input is joined to a continuous one, weigh the maximum policy honestly against the records it will discard.

Diagnosing state growth from query progress

Every completed micro-batch emits a progress record: query.lastProgress from a shell, query.recentProgress for the recent history, a StreamingQueryListener for anything you intend to keep. A handful of its fields answer nearly every watermark question you will have.

{
  "batchId" : 4181,
  "numInputRows" : 182044,
  "durationMs" : { "addBatch" : 21455, "triggerExecution" : 24012 },
  "eventTime" : {
    "max"        : "2026-08-06T14:52:31.000Z",
    "watermark"  : "2026-08-06T14:42:31.000Z"
  },
  "stateOperators" : [ {
    "numRowsTotal"              : 41850233,
    "numRowsUpdated"            : 96114,
    "numRowsRemoved"            : 0,
    "numRowsDroppedByWatermark" : 0,
    "memoryUsedBytes"           : 9884213760
  } ]
}

eventTime.watermark against eventTime.max

In a single-input query the gap between them should equal your delay threshold exactly - in the sample above, ten minutes. With several inputs under the default minimum policy the reported global watermark is set by whichever input is furthest behind, so expect the gap against this operator's own maximum to be wider than the threshold, and read a widening gap as a lagging sibling stream rather than a misconfigured delay. A null watermark means the query has none, so no state will ever be released. A watermark that holds the same value across consecutive batches while numInputRows stays healthy is the idle-input case: either a source has stopped, or the minimum policy is pinning you to one that has.

stateOperators and numRowsTotal

This is the number to alert on. Healthy state is flat or sawtoothed - it builds through a window and falls when eviction runs. Monotonic growth across hours means eviction is never firing, and there are only a few causes: complete output mode, a withWatermark that is missing or attached to the wrong column, a stream-stream join with no range condition, or a grouping key whose cardinality is genuinely unbounded, such as a raw session id or a URL with its query string attached. The array carries one entry per stateful operator in plan order, which is how you tell a runaway join from a runaway aggregation inside the same query. Note the zero in numRowsRemoved above alongside forty-one million rows held - that combination is the signature.

numRowsDroppedByWatermark

Anything above zero means real records were thrown away for being late. A small steady trickle is the trade you consciously made when you chose the threshold. A step change almost always means a producer started lagging. A large value alongside a healthy-looking watermark is the waking-partition backlog from the previous section, and it will not appear anywhere else in your monitoring.

durationMs and memoryUsedBytes

When addBatch starts to dominate triggerExecution, and memoryUsedBytes tracks numRowsTotal upward, and triggerExecution creeps toward the trigger interval, the query is on its way to falling permanently behind its source. The fix is upstream of the state store every time: shorten the delay, coarsen the window, or cut key cardinality. Raising spark.sql.shuffle.partitions spreads the same state across more store instances and can help with skew, but it does not reduce the number of rows you are obliged to keep.