Choosing a Watermark Delay for Spark Structured Streaming State

The problem: late events disappear, or state never shrinks

You run a Structured Streaming job that aggregates click events into five-minute windows. The dashboard looks right during a quiet test, but in production two things happen: session counts are lower than expected, and executor memory climbs until the task is killed. The usual culprit is the event-time watermark delay passed to withWatermark. Set it too short and Spark drops events that arrive a few minutes late. Set it too long and Spark keeps aggregation state for hours, bloating the state store and slowing recovery after a restart.

What the watermark controls

In Structured Streaming, event time is the timestamp embedded in the record itself, not the time Spark reads it. A watermark is Spark’s moving threshold for how old an event can be and still update state. For a column eventTime and delay delayThreshold, Spark computes the watermark as roughly the maximum event time seen so far minus the delay. Stateful operations—windowed aggregations, stream-stream joins, and deduplication—drop state older than that threshold. Any record arriving after its event time has fallen behind the watermark is ignored by those operations.

Watermarks are not a delivery guarantee. They are a state-retention policy. An event that arrives late is not queued or retried; it is excluded from the running result.

Choosing the delay from observed lateness

Start by measuring how long after the event time your records actually arrive. If your source records include processing time or ingestion time, subtract event time from it and plot the distribution. A practical starting point is the 95th or 99th percentile of that lag, plus a safety margin for clock skew and network pauses.

Delay setting	Effect on late data	Effect on state and recovery
Too short (for example, 1 minute)	Drops events delayed by minutes or hours	Small state store, fast checkpoint cleanup
Matched to observed p99 lag	Retains most legitimate late events	Moderate state, predictable recovery
Very long (for example, 24 hours)	Almost no late events dropped	Large state store, slow restarts, larger checkpoints

Worked example: configure and verify

Run this in a Spark shell, notebook, or spark-submit job. You need read access to the source path and write access to the checkpoint directory.

import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("session-aggregation")
  .getOrCreate()

spark.sparkContext.setLogLevel("WARN")

val raw = spark.readStream
  .format("kafka")
  .option("subscribe", "")
  .option("startingOffsets", "latest")
  .load()
  .selectExpr("CAST(value AS STRING) as json")
  .select(from_json(col("json"), schema).as("data"))
  .select(col("data.userId"), col("data.eventTime").cast("timestamp"))

val withWM = raw
  .withWatermark("eventTime", "10 minutes")

val sessions = withWM
  .groupBy(
    window(col("eventTime"), "5 minutes"),
    col("userId")
  )
  .count()

val query = sessions.writeStream
  .format("console")
  .outputMode("update")
  .option("checkpointLocation", "")
  .start()

query.awaitTermination()

Replace and with your own values. The checkpoint directory must be writable by the Spark executor user; on a shared cluster, choose a path under a directory owned by that user or grant write permission before starting the query.

Verification steps

1. Start the query and confirm the console sink prints aggregation rows.
2. Send a test event whose eventTime is older than the current watermark—for example, 12 minutes behind the latest event time when the delay is 10 minutes.
3. Check that the event does not change the running count for its window. If it does, the watermark may not have advanced as expected or the delay is too short.
4. Open the Spark UI Streaming tab while the query runs. Inspect state store metrics and the checkpoint directory size. Growing state without bound suggests the delay is too long or the data contains ever-increasing event times with few gaps.

Do not treat a single console run as proof. Repeat the test with several late-event offsets and compare against the Spark version you plan to deploy, because watermark and state-store defaults change across major releases.

Limitation: late data is excluded, not delayed

A watermark does not hold results open indefinitely for stragglers. Once an event’s event time is older than the watermark, stateful operators drop it. In stream-stream joins, Spark may retain state longer than the configured delay because watermarks propagate conservatively across both sides of the join. If your business requires counting every late event, consider writing raw events to immutable storage and running a separate batch correction job rather than relying only on streaming state.

Actionable closing

1. Measure event arrival lag in your source system and pick a delay near the observed p99, not an arbitrary default.
2. Configure withWatermark on the streaming DataFrame before any stateful operation.
3. Verify with console output and Spark UI state-store metrics before promoting to production.
4. Recheck the delay after source schema changes, clock-migration events, or Spark version upgrades.

The right watermark is the smallest delay that retains the late events your downstream consumers actually need.