In Spark Structured Streaming, what does withWatermark do to state and to late rows?

## Why a watermark exists at all A streaming aggregation has to answer a question a batch job never faces: *when is a group finished?* A batch `GROUP BY` reads the whole input and then emits, but a stream has no end. If Spark kept every window's partial aggregate forever, state would grow without bound; if it closed windows eagerly, records arriving a little late would be lost or would produce a second, contradictory answer. The watermark is the explicit lateness contract that resolves this. `withWatermark(colName, delayThreshold)` says: *for this event-time column, I am willing to wait `delayThreshold` past the newest event time I have seen; beyond that, I accept losing records.* Everything else — when windows emit, when state is released, which rows are dropped — follows from that one declaration. ## How the value is computed The watermark is derived from the data, not from the wall clock. At the end of a micro-batch Spark takes the maximum value of the watermarked event-time column across every row it has processed so far and subtracts the threshold. That value is then used by stateful operators during the *following* micro-batch, so the watermark always trails the data by one trigger. Two consequences matter in practice. First, the watermark only ever moves forward — it is monotonic, so a burst of very old records cannot drag it backwards. Second, it advances only from data the query actually reads: if the stream goes quiet, the maximum event time stops moving, the watermark freezes, and buffered windows never close. A stalled watermark on an idle source is one of the most common "my streaming job stopped producing output" reports. ## Where to call it `withWatermark` must be applied to the event-time column *before* the stateful operator that should use it, and the same column has to be the one the aggregation groups on (directly, or via `window()`), otherwise the watermark has nothing to bind to. Applying it after the aggregation, or on a different column than the one being windowed, produces a query where the watermark simply has no effect — the query runs, state grows, and nothing is ever dropped or evicted. ## What it does for each operator **Windowed aggregations.** A window is finalized when the watermark passes the window's end. In append mode the window's row is emitted at that moment and its state is released; in update mode revisions are emitted continuously and the state is released at the same point. Effective output latency is therefore roughly *window length + delay threshold*. **Arbitrary stateful processing.** `flatMapGroupsWithState` with `GroupStateTimeout.EventTimeTimeout()` fires a group's timeout callback when the watermark passes the timestamp you registered, which is how you expire sessions. **Stream–stream joins.** Watermarks on both sides plus a time-range condition on the join let Spark bound how long unmatched rows must be buffered. Outer joins require this, because the NULL-padded row can only be emitted once the watermark proves no match will ever arrive. **Deduplication.** `dropDuplicates` on a watermarked stream keeps the seen-keys set bounded by the watermark instead of forever. ## Multiple input streams When a query reads more than one stream, each source contributes its own watermark and Spark reduces them to a single global value. The default policy is the **minimum**, controlled by `spark.sql.streaming.multipleWatermarkPolicy` (`min` by default, `max` available). Taking the minimum is the safe choice — it means a slow or backfilling source holds the whole query's watermark back rather than causing its counterpart's records to be dropped. It also means one lagging input can stall output for everything downstream, which is worth knowing before you union a live topic with a historical replay. ## The guarantee is asymmetric Spark is careful about what it promises. A record delayed by less than the threshold **is** processed. A record delayed by more than the threshold **may or may not** be dropped — the more delayed it is, the more likely it is discarded, but there is no promise that a slightly-over-threshold record is thrown away. Do not build correctness on "anything late is definitely dropped"; build on "anything within the threshold is definitely counted." ## Observing it `query.lastProgress` exposes an `eventTime` block containing the current `watermark`, and each entry in `stateOperators` reports `numRowsTotal`, `numRowsUpdated` and `numRowsDroppedByWatermark`. Those three numbers answer most watermark questions in production: whether the watermark is advancing at all, how big state has grown, and how much data you are actually losing to lateness. A steadily climbing `numRowsTotal` with a frozen `watermark` is the signature of a watermark that is not doing its job. ## Choosing the threshold The threshold is a latency-versus-completeness dial. Larger means fewer dropped records, more retained state, and later output in append mode; smaller means faster, cheaper, lossier. Size it from a measured distribution of *(arrival time − event time)* on the real source rather than from intuition — a 99th-percentile lag with a margin is the usual starting point.