Event-Time Watermarking

In real-world streaming pipelines (like web clickstreams or IoT sensors), data can arrive late at the Spark cluster due to network delays, device outages, or high latency. To aggregate data accurately, Spark must calculate metrics based on Event-Time (the timestamp when the event actually occurred at the device) rather than the arrival time at Spark.

To prevent infinite RAM usage when tracking historical event states, Spark utilizes Watermarking.

Tumbling vs. Sliding Windows

When aggregating streaming data over time, Spark groups records into specific time blocks:

Tumbling Windows (Non-Overlapping):
[ 12:00 - 12:10 ]  [ 12:10 - 12:20 ]  [ 12:20 - 12:30 ]

Sliding Windows (Overlapping):
[ 12:00 - 12:10 ]
      [ 12:05 - 12:15 ]
            [ 12:10 - 12:20 ]

• Tumbling Window: Fixed, non-overlapping intervals (e.g. 10-minute blocks). A record falls into exactly one window.
• Sliding Window: Overlapping intervals (e.g. 10-minute window sliding every 5 minutes). A record can fall into multiple windows.

What is Watermarking?

A Watermark is a dynamic time boundary that tells Spark how late late-arriving data can be before it is ignored.

df.withWatermark("event_time", "10 minutes")

How it Works:

1. Spark monitors the maximum event time seen so far in the stream (e.g., 12:30).
2. The watermark is calculated as: Max Event Time - Threshold (e.g., 12:30 - 10 minutes = 12:20).
3. Any late-arriving data with an event timestamp older than 12:20 is immediately dropped and ignored.
4. State Clean Up: Crucially, Spark completely clears the aggregated window state for windows older than 12:20 from the executor's memory, ensuring that your streaming job's memory footprint does not grow infinitely!

PySpark Code Example: Windowing with Watermarks

Here is a complete script demonstrating event-time aggregations using a sliding window and a 10-minute watermark:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# 1. Setup Spark
spark = SparkSession.builder \
    .appName("Streaming Watermarks") \
    .master("local[*]") \
    .getOrCreate()

# 2. Ingest sensor data stream
sensor_stream_df = spark.readStream \
    .format("json") \
    .schema("device_id STRING, event_time TIMESTAMP, reading DOUBLE") \
    .load("sensor_stream_directory")

# 3. Define Watermark & Aggregations
# We partition data into 10-minute windows sliding every 5 minutes,
# allowing data to arrive up to 10 minutes late.
aggregated_df = sensor_stream_df \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(
        F.window(F.col("event_time"), "10 minutes", "5 minutes"),
        F.col("device_id")
    ) \
    .agg(F.avg("reading").alias("avg_reading"))

# 4. Write stream to Console in UPDATE mode
# Update mode is standard for watermarked aggregations
query = aggregated_df.writeStream \
    .format("console") \
    .outputMode("update") \
    .option("checkpointLocation", "watermark_checkpoints") \
    .start()

query.awaitTermination()