Event-Time, Windows, Watermarks & Stateful Streaming

In "Stream Processing" you learned the ideas: event time, windows, watermarks and state. This chapter shows the Spark code for each idea, and the rules Spark adds.

"Structured Streaming Fundamentals" came first. It explained micro-batches, output modes and the checkpoint. Here you use them.

The data is the shop stream again, on 2026-01-22. Spark reads the orders in the order they arrive, in micro-batches 0, 1 and 2. Order 113 is late: it happened at 10:04 and arrives at 10:07.

The chapter has ten parts.

1. Event time and tumbling windows.
2. Sliding windows.
3. Session windows.
4. Watermarks and state.
5. Late rows.
6. Joining two streams.
7. Removing duplicates.
8. Your own state, with timeouts.
9. The state store and how to monitor it.
10. A recap with questions.

Event time is when something happened. It is a column in the data. Here it is order_ts, the time the customer placed the order.

A window is a block of event time, such as 10:00 to 10:05. The window starts at 10:00 and stops just before 10:05. A tumbling window block never overlaps the next one.

The function window() puts each row into its block. Look at a count and a total for each 5-minute window, with no watermark yet.

from pyspark.sql.functions import window, count, sum as sum_

by_window = (orders
    .groupBy(window("order_ts", "5 minutes"))
    .agg(count("*").alias("orders"),
         sum_("amount").alias("total")))

# written with .outputMode("update")

Update mode prints only the windows that changed. In batch 2, the 10:00 window did not change, so it is missing. Order 151 at exactly 10:10 starts a new window.

Spark must remember the count and total of every window across batches. This remembered data is called state. It is kept for you, and it is saved in the checkpoint.

A sliding window has a length and a slide. A 10-minute window with a 5-minute slide starts every 5 minutes. Each window is 10 minutes long, so windows overlap.

One row falls into two windows. Look at orders 110 to 114 with this query.

by_window = (orders
    .groupBy(window("order_ts", "10 minutes", "5 minutes"))
    .agg(count("*").alias("orders"),
         sum_("amount").alias("total")))

Use a sliding window for a moving number, such as "orders in the last 10 minutes, updated every 5 minutes".

A session window has no fixed size. It grows while a customer keeps ordering, and it ends after a quiet gap. Spark 3.2 added session_window for this.

Look at customer 2 with a gap of 3 minutes. Order 190 is a made-up extra order at 10:04.

from pyspark.sql.functions import session_window

sessions = (orders
    .groupBy("customer_id", session_window("order_ts", "3 minutes"))
    .agg(count("*").alias("orders"),
         sum_("amount").alias("total")))

A late row can fill the gap between two sessions. Then Spark joins them into one. For this reason, Spark can only call a session final when the watermark reaches its end.

A watermark is a cut-off in event time. It is the newest event time Spark has seen, minus a delay. A window that ends at or before the cut-off is final.

You set it with withWatermark(column, delay). We use a delay of 3 minutes, the same as in "Stream Processing". Look at the same window query, in append mode.

by_window = (orders
    .withWatermark("order_ts", "3 minutes")
    .groupBy(window("order_ts", "5 minutes"))
    .agg(count("*").alias("orders"),
         sum_("amount").alias("total")))

query = (by_window.writeStream
    .outputMode("append")
    .format("console")
    .option("checkpointLocation", "/chk/by-window")
    .start())

Follow these four rules.

1. Spark works out the watermark at the end of a batch, and uses it in the next batch. A batch never uses its own rows for its own cut-off.
2. Call withWatermark before the aggregation, on the same column that the window uses. Otherwise Spark does not use it to clear state.
3. The watermark never goes down. A late row cannot move it back.
4. Append mode prints a window once, after it is final. Update mode prints a window every time it changes, and uses the watermark to drop old state and too-late rows.

Two rows are late in the shop stream. Order 113 arrives in batch 1. Order 160 arrives in batch 3.

Order 160 is the very late order from "Stream Processing". It was placed at 10:03 and arrives alone, at offset 7, in a new batch. Look at what the watermark does to each late row.

# same query: withWatermark("order_ts", "3 minutes"),
# window("order_ts", "5 minutes"), append mode

Spark gives one promise. Rows that are no more than the delay behind the newest event time are kept. Spark does not promise to drop older rows.

An older row can still be counted if its window is still in state. Never write logic that depends on a row being dropped.

Pick the delay with care. A longer delay catches more late rows, but holds more state and writes append results later. A shorter delay does the opposite.

A stream-stream join matches rows from two streams. Say a second stream holds payments. Spark must keep rows in state while a match can still arrive.

You give Spark a time rule, so it knows when to stop waiting. Here a payment must come at most 5 minutes after the order. Both payments are made up.

from pyspark.sql.functions import expr

o = orders.withWatermark("order_ts", "3 minutes").alias("o")
p = payments.withWatermark("paid_ts", "3 minutes").alias("p")

paid = o.join(p, expr("""
    o.order_id = p.order_id
    AND p.paid_ts >= o.order_ts
    AND p.paid_ts <= o.order_ts + interval 5 minutes"""), "inner")

With "left_outer", Spark writes order 111 with NULL only after the watermark passes 10:09, which is 10:04 plus 5 minutes. Until then, a payment could still come.

Here is what each join type needs.

A query has one global watermark. With two streams, Spark uses the smaller value by default.

The setting spark.sql.streaming.multipleWatermarkPolicy controls this. Its default is min.

A source can send the same order twice, for example after a retry. You met this in "Delivery Guarantees & Ordering". In Spark, you remove duplicates with dropDuplicates.

Look at three rows. Order 110 arrives twice, with the same event time.

deduped = (orders
    .withWatermark("order_ts", "3 minutes")
    .dropDuplicates(["order_id", "order_ts"]))

There are three ways to do it.

The third way needs Spark 3.5 or later. Use it when a retry can change the event time.

Sometimes a built-in aggregation cannot do the job. You then write a function that keeps its own state for each key. Spark calls it for every key that has new rows.

In PySpark 3.4 and later, the method is applyInPandasWithState. In Scala and Java, use flatMapGroupsWithState or mapGroupsWithState.

Look at one example. It counts how many orders of 10.00 or more each customer placed in a row. A smaller order sets the count back to 0.

import pandas as pd
from pyspark.sql.streaming.state import GroupStateTimeout

def streak(key, pdf_iter, state):
    n = state.get[0] if state.exists else 0
    for pdf in pdf_iter:
        for amount in pdf["amount"]:
            n = n + 1 if amount >= 10 else 0
    state.update((n,))
    yield pd.DataFrame({"customer_id": [key[0]], "streak": [n]})

streaks = orders.groupBy("customer_id").applyInPandasWithState(
    streak,
    outputStructType="customer_id INT, streak LONG",
    stateStructType="streak LONG",
    outputMode="Update",
    timeoutConf=GroupStateTimeout.NoTimeout)

The function reads the old streak with state.get and saves the new one with state.update. Spark stores it in the state store and in the checkpoint.

A timeout lets you act when a key goes quiet. With a processing-time timeout, you set timeoutConf to GroupStateTimeout.ProcessingTimeTimeout and call state.setTimeoutDuration. Later, Spark calls your function with no rows, and state.hasTimedOut is true.

All state lives in the state store. By default, it keeps the state in memory on the executors, and saves copies in the checkpoint folder.

Large state can fill the memory. Spark 3.2 added the RocksDB provider, which keeps state on local disk. You turn it on with one setting, before the first run of a query.

spark.conf.set(
    "spark.sql.streaming.stateStore.providerClass",
    "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

To watch state, read query.lastProgress. Look at a shortened report for batch 2 of the append query. The memory number is made up.

query.lastProgress
# {
#   "batchId": 2,
#   "numInputRows": 2,
#   "eventTime": {"max": "2026-01-22T10:10:00.000Z",
#                 "min": "2026-01-22T10:09:00.000Z",
#                 "avg": "2026-01-22T10:09:30.000Z",
#                 "watermark": "2026-01-22T10:05:00.000Z"},
#   "stateOperators": [{"numRowsTotal": 2, "memoryUsedBytes": 21480}]
# }

Five points to keep.

• window() groups rows by event time. Tumbling windows do not overlap. Sliding windows do. session_window grows with activity.
• The watermark is the newest event time minus the delay. Spark uses it in the next batch, and it never goes down.
• Append mode writes a window once, after it is final. Update mode writes every change.
• Spark keeps rows inside the delay. It does not promise to drop older rows.
• All state needs a way out: a watermark, or state.remove in your own function. Otherwise it grows forever.

Answer these without looking back.

1. Order 113 has event time 10:04 and arrives in batch 1. Why is it kept?
2. What watermark does batch 2 use in our query, and why?
3. When does append mode write the window 10:05 to 10:10?
4. Why does a left outer join need a watermark and a time rule?
5. What is the difference between dropDuplicates and dropDuplicatesWithinWatermark?