Definition
- Definition – A watermark is a moving threshold on event time computed as the maximum event time observed so far minus a configured delay, declared with
withWatermark("eventTime", "10 minutes"). - Late-data handling – Events with an event time older than the current watermark are considered too late and dropped from aggregation.
- State cleanup – State held for windows that have fully closed below the watermark is evicted, which is what bounds memory in long-running stateful queries.
- Monotonic advance – The watermark only advances forward; it never moves backward even if older events arrive.
- Essential for stateful queries – Without a watermark, windowed aggregations, stream-stream joins, and deduplication would keep state for every open window forever.
At the National Bank of India the central office must decide how long to keep an hour's tally open for late-arriving slips. The watermark is the rule: "we will wait up to ten minutes past the latest slip we have seen; anything older than that is too late and will not change closed tallies." Once the clock on the newest slip advances far enough, the office finalizes and files away the old hour's tally and stops holding paperwork for it. Without such a rule the office would have to keep every hour's tally open forever.
Understanding the Concepts
What is a watermark?
A watermark in Structured Streaming is a moving threshold on event time that tells Spark how late an event may arrive before it is considered too old to process. You declare it before any aggregation with withWatermark("eventTime", "10 minutes"), and Spark continuously computes the watermark as the maximum event time seen so far minus the configured delay.
import static org.apache.spark.sql.functions.*;
Dataset<Row> transactions = spark.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "broker1:9092")
.option("subscribe", "nbi-transactions")
.load()
.select(from_json(col("value").cast("string"), txnSchema).as("txn"))
.select("txn.*");
// Declare a 10-minute watermark on the event-time column
Dataset<Row> withWatermark = transactions
.withWatermark("eventTime", "10 minutes");
How is the watermark computed?
The watermark is computed from the data, not the wall clock. Spark tracks the maximum event-time value it has observed across all partitions of the stream, and the current watermark equals that maximum minus the delay you configured. The watermark advances monotonically and never moves backward.
// If the latest event seen has eventTime = 11:20
// Watermark = 11:20 - 10 minutes = 11:10
// Events with eventTime < 11:10 are dropped as too late
// Next batch: latest eventTime = 11:25
// Watermark = 11:25 - 10 minutes = 11:15 (advanced)
// Now events with eventTime < 11:15 are dropped
Batch 0: max eventTime = 11:20 → watermark = 11:10
Batch 1: max eventTime = 11:25 → watermark = 11:15
Batch 2: max eventTime = 11:23 → watermark = 11:15 (no retreat)
(watermark stays at 11:15 because 11:23 < 11:25)
Why is the watermark essential for bounding state?
In a stateful streaming query such as a windowed aggregation, Spark must hold state for every open window. Without a watermark, state grows without bound because in principle any future event might belong to an old window. The watermark declares a point past which late data is ignored, so Spark can finalize and release old state.
// Without watermark -- state grows forever
Dataset<Row> unbounded = transactions
.groupBy(window(col("eventTime"), "5 minutes"), col("accountId"))
.agg(sum("amount").as("total"));
// State for every 5-minute window is kept indefinitely!
// With watermark -- state is bounded
Dataset<Row> bounded = transactions
.withWatermark("eventTime", "10 minutes")
.groupBy(window(col("eventTime"), "5 minutes"), col("accountId"))
.agg(sum("amount").as("total"));
// State for windows that end before (max_event_time - 10min) is evicted
How does the watermark interact with output modes?
In Append mode with aggregation, a window's result is emitted only once the watermark passes the end of the window, because Append requires that an emitted row is final. In Update mode, intermediate results can be emitted before finalization, but the watermark still governs when state is dropped. This is why Append aggregation without a watermark is rejected at query start.
// Append mode -- result emitted ONLY when watermark passes window end
transactions
.withWatermark("eventTime", "10 minutes")
.groupBy(window(col("eventTime"), "5 minutes"), col("accountId"))
.count()
.writeStream()
.outputMode("append") // row emitted once, guaranteed final
.start();
// Update mode -- intermediate results emitted, watermark cleans state
transactions
.withWatermark("eventTime", "10 minutes")
.groupBy(window(col("eventTime"), "5 minutes"), col("accountId"))
.count()
.writeStream()
.outputMode("update") // changed rows emitted each batch
.start();
What operations require a watermark?
Watermarks are required or strongly recommended for windowed aggregations, stream-stream joins, and dropDuplicates with an event-time column. Each of these stateful operations would otherwise hold state indefinitely.
// Windowed aggregation -- watermark required for Append mode
transactions.withWatermark("eventTime", "10 minutes")
.groupBy(window(col("eventTime"), "5 minutes"))
.count();
// Stream-stream join -- watermark on both sides
Dataset<Row> orders = orderStream.withWatermark("orderTime", "10 minutes");
Dataset<Row> payments = paymentStream.withWatermark("payTime", "10 minutes");
orders.join(payments,
expr("orderId = payOrderId AND " +
"payTime >= orderTime AND " +
"payTime <= orderTime + interval 1 hour"));
// Deduplication -- watermark bounds how long keys are remembered
transactions.withWatermark("eventTime", "10 minutes")
.dropDuplicates("transactionId", "eventTime");
What is the tuning trade-off?
A larger watermark delay tolerates later events and improves completeness but holds state longer and delays finalization, raising memory and latency. A smaller delay finalizes quickly and uses less memory but silently drops more late events. In banking you size the delay to the real lateness distribution of the feed.
// Tight watermark -- fast finalization, may drop stragglers
transactions.withWatermark("eventTime", "30 seconds");
// Windows finalize quickly, low memory
// But a transaction delayed 2 minutes is dropped
// Relaxed watermark -- captures more, higher latency
transactions.withWatermark("eventTime", "30 minutes");
// Windows stay open 30 extra minutes
// Captures almost all late events, but uses more memory
What are common mistakes?
The most common mistake is forgetting the watermark entirely and having state grow without bound. Another mistake is declaring the watermark on a column that is not the same event-time column used in the window function, which makes the watermark ineffective.
// WRONG -- watermark on different column than window
transactions
.withWatermark("ingestTime", "10 minutes") // watermark on ingestTime
.groupBy(window(col("eventTime"), "5 minutes")) // window on eventTime
.count();
// Watermark does not bound the window state!
// CORRECT -- same column for both
transactions
.withWatermark("eventTime", "10 minutes")
.groupBy(window(col("eventTime"), "5 minutes"))
.count();
What about monitoring dropped late events?
In production you should monitor the number of dropped late events to tell whether the watermark is too tight. If the dropped count is rising, you may need to increase the watermark delay. If it is always zero, you could potentially tighten it to reduce state.
// Monitor via StreamingQueryListener
spark.streams().addListener(new StreamingQueryListener() {
@Override
public void onQueryProgress(QueryProgressEvent event) {
// Check stateOperators for rows dropped
System.out.println("State rows total: " +
event.progress().stateOperators()[0].numRowsTotal());
System.out.println("State rows dropped: " +
event.progress().stateOperators()[0].numRowsDroppedByWatermark());
}
// ... other overrides
});
What should a senior engineer know?
The watermark is a deliberate correctness-versus-resource control, not a default you set and forget. You size the delay to the real lateness distribution of your feed, often covering roughly the 99th-percentile observed delay, and you monitor the dropped-late metric so you can tell whether the watermark is too tight. In banking, an undercounted transaction total is a compliance problem, so the watermark must be wide enough to capture the vast majority of late events while still keeping state bounded.
Visual Overview
Watermark Computation and Effect
Stream of events with event times:
┌────────┬────────┬────────┬────────┬────────┐
│ 11:15 │ 11:18 │ 11:20 │ 11:12 │ 11:22 │
└────────┴────────┴────────┴────────┴────────┘
▲ late!
Watermark delay: 10 minutes
After seeing 11:20:
max(eventTime) = 11:20
watermark = 11:20 - 10 min = 11:10
Event 11:12 → 11:12 > 11:10 → ACCEPTED
After seeing 11:22:
max(eventTime) = 11:22
watermark = 11:22 - 10 min = 11:12
Any future event < 11:12 → DROPPED
State Cleanup:
┌──────────────┬──────────────┬──────────────┐
│ Window │ Window │ Window │
│ [11:00,11:05)│ [11:05,11:10)│ [11:10,11:15)│
│ State: EVICT │ State: EVICT │ State: KEPT │
│ (end < WM) │ (end < WM) │ (end > WM) │
└──────────────┴──────────────┴──────────────┘
▲ ▲
Watermark = 11:12
Windows ending before 11:12 have state cleaned up
Key Takeaway
A watermark is a moving threshold on event time that tells Spark how late an event may arrive before it is treated as too old. You declare it with withWatermark on the event-time column and an allowed-lateness delay, and Spark computes the current watermark as the maximum event time it has seen minus that delay. The watermark only advances, never moves backward. Two things happen against it: any event older than the current watermark is dropped from the aggregation, and the state for any window that ends before the watermark is cleaned up because no further events can change it.
That second effect is why the watermark is essential for stateful streaming. In a windowed aggregation, a stream-stream join, or a deduplication, Spark must hold state, and without a watermark it would have to keep every window's state forever because in principle any future event could belong to an old window. The watermark declares a point past which late data is ignored, so Spark can finalize and release old state and keep memory bounded over a long-running job.
It also pairs with output mode: in Append mode a windowed result is emitted only once the watermark passes the window's end, which is why Append aggregation without a watermark is rejected at query start. In Update mode the watermark still governs state eviction even though intermediate results are emitted earlier.
At the National Bank of India Rajesh used a ten-minute watermark on a per-minute windowed transaction count. It let the job wait ten minutes for stragglers, then finalized and dropped each minute's state so memory stayed flat over days of running. He sized the delay to roughly the 99th-percentile observed lateness of the feed and monitored the dropped-late metric to detect if the watermark was too tight.
Remember one sentence: The watermark is max-event-time-seen minus allowed lateness, it drops too-late events and evicts closed-window state, and it is the one mechanism that keeps stateful streaming state from growing without bound.
---