Abstract

Watermarks in Spark Structured Streaming define the boundary between data the engine is still willing to process and data it considers too late to include in results. Getting watermark configuration right is one of the most consequential tuning decisions in a Structured Streaming deployment - too aggressive a watermark drops legitimate late events; too conservative a watermark forces the engine to maintain unbounded state, eventually exhausting memory. Understanding how Spark tracks event time, advances its internal watermark threshold, and decides when to emit window results is essential for operators who need correct aggregations over out-of-order event streams.

How Watermarks Advance

When a streaming query includes withWatermark("event_time", "10 minutes"), Spark tracks the maximum event timestamp observed across all partitions in each micro-batch. At batch end, the threshold updates to max_observed_time - 10 minutes. Events below the threshold are silently dropped.

The critical subtlety: the watermark advances globally, not per partition. A single slow partition holding an old timestamp stalls cleanup for all partitions. Monitoring the gap between wall-clock time and watermark time (via StreamingQuery.lastProgress) is essential. A gap exceeding twice the configured delay is a signal that a partition is stalled or receiving extremely late data.

Window Types and Result Emission

Structured Streaming supports tumbling, sliding, and session windows (session windows added in Spark 3.2). For tumbling and sliding windows, the engine emits a final result only after the watermark advances past the window’s end time plus the configured delay - results trail real time by the full delay amount.

outputMode controls sink writes: append emits only finalized rows (correct for watermarked windows, since late data was discarded), update emits aggregates per batch (requires upsert-capable sinks), and complete re-emits the full aggregation table each batch (only practical for small aggregations).

State Store and Tuning

Each unique grouping key occupies state in the RocksDB-backed or in-memory store. Without proper watermark tuning, state grows without bound. Spark’s state store metrics - numRowsTotal and memoryUsedBytes per operator, accessible via the Structured Streaming UI or Prometheus scraping - signal health: a stable numRowsTotal at steady state means cleanup is working; a growing value means the watermark is not advancing fast enough.

For high-cardinality groupings, the RocksDB state store (set via spark.sql.streaming.stateStore.providerClass) is strongly preferred. It spills to disk and supports state sizes exceeding JVM heap. Databricks Runtime ships tuned RocksDB configurations that reduce compaction overhead for typical streaming workloads.

Set the watermark delay to the 95th-to-99th percentile of observed late-arrival lag, not the maximum. Designing for the maximum extends state retention unnecessarily. When joining two streaming sources, each requires its own watermark, and Spark waits for both to advance before emitting join results - this compounds latency and requires coordinated delay settings across both input streams.