Spark Streaming Windows and Watermarks

Streaming data arrives without end, so questions such as "how many clicks in the last five minutes?" need a time window. A window groups events by time. A watermark tells Spark how long to wait for late events before it closes a window and frees memory.

The Layman View

A bus company counts passengers per hour. Some tap cards arrive late because of weak network signals. The company waits ten extra minutes before it publishes the hourly number. After that deadline the hour is final. A watermark sets that deadline.

Event Time and Processing Time

• Event time marks when the event happened, as recorded in the data.
• Processing time marks when Spark received the event.

Network delays make the two times differ. Event time gives correct answers. Spark windows on event time when you group by a timestamp column.

Tumbling Windows

A tumbling window cuts time into equal blocks that never overlap. Each event belongs to exactly one block.

from pyspark.sql import functions as F

counts = (events
    .groupBy(F.window("event_time", "10 minutes"))
    .count())

Time:   10:00        10:10        10:20        10:30
        |------------|------------|------------|
        Window A     Window B     Window C

Sliding Windows

A sliding window moves forward in smaller steps than its length. The windows overlap, and one event can fall into several windows.

counts = (events
    .groupBy(F.window("event_time", "10 minutes", "5 minutes"))
    .count())

Time:   10:00   10:05   10:10   10:15
        |----------------|
                |----------------|
                        |----------------|

The Late Data Problem

An event stamped 10:07 may reach Spark at 10:16. Spark keeps the 10:00 window open to accept it. Open windows consume memory. Without a limit, Spark would keep every window forever.

Adding a Watermark

The withWatermark call names the event time column and the allowed lateness.

counts = (events
    .withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "10 minutes"))
    .count())

Spark tracks the newest event time seen so far. The watermark equals that newest time minus the allowed lateness.

Newest event time seen : 10:25
Allowed lateness       : 10 minutes
Watermark              : 10:15

Window 10:00 - 10:10 ends before 10:15 --> closed, state dropped
Window 10:10 - 10:20 ends after  10:15 --> still open

Late event stamped 10:07 arrives --> ignored (window closed)
Late event stamped 10:12 arrives --> accepted (window open)

Output Modes

• append emits a window once, after the watermark passes its end. The result is final.
• update emits windows each time their counts change.
• complete emits the full result table every time. It keeps all state and does not suit long runs.

Complete Example

stream = (spark.readStream.format("rate").option("rowsPerSecond", 5).load()
            .withColumnRenamed("timestamp", "event_time"))

query = (stream
    .withWatermark("event_time", "1 minute")
    .groupBy(F.window("event_time", "30 seconds"))
    .count()
    .writeStream
    .outputMode("append")
    .format("console")
    .option("truncate", False)
    .start())

query.awaitTermination()

The rate source generates test rows with timestamps, so no external system is needed.

Choosing the Lateness Value

• Measure the real delay in your data. Look at the gap between event time and arrival time.
• A larger watermark accepts more late events but holds more state in memory and delays append output.
• A smaller watermark saves memory and risks dropping valid late events.

Tips

• Use the same column in withWatermark and in the window.
• Call withWatermark before the aggregation.
• Store checkpoints in reliable storage so state survives restarts.