advanced
Watermarks
10 min readLast updated: 2026-07-09
Overview
Learn how to use Watermarks to handle late-arriving records in streaming applications and clean up historical state store data.
What You Will Learn
In this lesson, you will learn:
- Late Data: Handling records that arrive out-of-order due to network delays.
- Watermark Threshold: Defining time thresholds for late data.
- State Store Cleanup: Purging historical aggregates from memory automatically.
Detailed Concept Explanation
In real-time systems, network delays or device offline states can cause events to arrive at Spark long after they actually occurred.
A Watermark defines a time threshold for how long Spark will wait for late-arriving data.
How it Works
- You specify a watermark using
.withWatermark("eventTimeColumn", "threshold")(e.g.10 minutes). - Spark tracks the maximum event time seen so far. The watermark boundary is:
Watermark = Max(Event Time) - Threshold - Any incoming record with an event time older than the watermark boundary is considered too late and is discarded immediately.
- The watermark also tells Spark when it is safe to purge historical aggregates from the state store, freeing up memory.
Code Examples
Python (PySpark) Implementation
python
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("Watermarks").getOrCreate()
# Stream sensor data
sensor_stream = spark.readStream.format("rate").load() \
.selectExpr("timestamp", "CAST(value AS STRING) AS device")
# Set 10-minute watermark on timestamp
watermarked_df = sensor_stream.withWatermark("timestamp", "10 minutes")
# Group by timestamp and device
grouped_df = watermarked_df.groupBy("timestamp", "device").count()
query = grouped_df.writeStream.format("console").start()
query.awaitTermination(5)
query.stop()
Expected Output
text
+-------------------+------+-----+
| timestamp|device|count|
+-------------------+------+-----+
|2026-07-09 19:30:00| dev1| 1|
+-------------------+------+-----+
Execution Plan Diagram (Python & Scala)
Execution Plan Diagram
SparkSession.builder
readStream
selectExpr
withWatermark(timestamp
10 minutes)
groupBy(timestamp
device).count()
writeStream.format(console)
start()
Scala Implementation
scala
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder().appName("WatermarksScala").getOrCreate()
val sensorStream = spark.readStream.format("rate").load()
.selectExpr("timestamp", "CAST(value AS STRING) AS device")
val watermarked = sensorStream.withWatermark("timestamp", "10 minutes")
val grouped = watermarked.groupBy("timestamp", "device").count()
val query = grouped.writeStream.format("console").start()
query.awaitTermination(5000)
query.stop()
Expected Output
text
+-------------------+------+-----+
| timestamp|device|count|
+-------------------+------+-----+
|2026-07-09 19:30:00| dev1| 1|
+-------------------+------+-----+
Common Mistakes
- Watermarking Non-Event-Time Columns: Attempting to set watermarks on system processing columns rather than the actual event timestamp. Watermarks must be applied to
TimestampTypecolumns containing event-time data.
Interview Perspective
What is watermarking in Structured Streaming and why is it necessary?
Watermarking is the mechanism used to handle late-arriving, out-of-order data in streaming queries. It sets a time threshold (e.g. 10 minutes) relative to the maximum event time observed. Data older than the threshold is discarded. It is necessary to prevent state stores from growing infinitely, as Spark uses the watermark to know when to safely purge old event aggregations from memory.