Spark Structured Streaming watermarks: windows by event time and late data
A watermark tells Spark how late an event can be and still count. withWatermark("event_time", "1 hour") before a windowed aggregation makes Spark track the newest event time it has seen, less 1 hour; a window that ends before that point is final. In append mode Spark writes each window once, when the watermark passes its end, and an event that arrives for a window already written is dropped. Choose the delay from how late your events really are, and recount from the raw table for the rare event later than that.
Module 9: Streaming
At the end of this part you can:
- Count a stream by event time in windows, with a watermark that bounds how long Spark waits for late events
- Predict when append mode writes a window, and which late events count and which are dropped
- Find the events a stream dropped as too late, and correct the totals from the raw table
Take this module into your AI
Open in ChatGPT (new tab)Open in Claude (new tab)Part 1 read the topic orders into bronze_orders. To start here, start Kafka and create the topic as part 1 does, save stream_session.py, make_orders.py, read_orders.py and show.py from it, and run python make_orders.py 1, python make_orders.py 2 and python read_orders.py.
2 clocks
Every event has 2 times: when it happened, its event time, and when the stream reads it. The shop wants orders per hour by when they were placed, so a till that loses its connection and sends 10 minutes of orders late must still have them counted in the hour they were placed. That raises the question every streaming count has to answer: when is an hour finished? An order from 09:50 could always still be on its way.
A watermark answers it with a rule. Spark tracks the newest event time it has seen and subtracts a delay you choose; an hour that ends before that point is final. Play with it here, then run it:
The newest is N, paid at 21:21
Time the payment was made (event time), in 5-minute windows
Latest payment seen, 21:21Watermark, 21:11Closed window
21:00
Final, state dropped
21:05
Final, state dropped
21:10
…
Waiting
21:15
…
Waiting
21:20
…
Waiting
Counted
13
Dropped as too late
1
Windows in memory
3
| Payment | Paid | Came after | Watermark | Result |
|---|---|---|---|---|
| D | 21:04 | 21:06 | before 21:00 | Counted |
| G | 21:02 | 21:12 | 21:02 | Counted |
| I | 21:08 | 21:14 | 21:04 | Counted |
| K | 21:13 | 21:17 | 21:07 | Counted |
| M | 21:03:30 | 21:19 | 21:09 | Dropped |
N, paid at 21:21, is the newest payment yet, so the watermark moves to 21:11. Waiting 10 minutes, 13 of 14 payments are counted and 3 windows are still held in memory. In append mode only the closed windows have been written.
The payments are made up, and each one is its own micro-batch. In Structured Streaming the watermark is the latest event time seen less the delay, and a window's state is dropped once the watermark passes its end. Spark promises to count anything less late than the delay; later data may or may not be counted.
Orders per hour
Save this as hourly_sales.py:
"""Add up the orders placed in each hour, by when they were placed, and write each hour once it is complete."""
from pyspark.sql import functions as F
from stream_session import stream_spark
spark = stream_spark("hourly_sales")
hourly = (
spark.readStream.table("bronze_orders")
.where("status = 'placed'")
# An event more than 1 hour older than the newest one seen is too late to count.
.withWatermark("event_time", "1 hour")
.groupBy(F.window("event_time", "1 hour"))
.agg(F.count("*").alias("orders"), F.sum("amount").alias("revenue"))
.select(F.col("window.start").alias("hour"), "orders", "revenue")
)
query = (
hourly.writeStream.format("delta")
.outputMode("append")
.option("checkpointLocation", "checkpoints/hourly_sales")
.trigger(availableNow=True)
.toTable("hourly_sales")
)
query.awaitTermination()
for progress in query.recentProgress:
dropped = sum(state.numRowsDroppedByWatermark for state in progress.stateOperators)
print(
f"batch {progress.batchId}: {progress.numInputRows} events in,",
f"{dropped} too late,",
f"watermark {progress.eventTime.get('watermark', 'none')}",
)
It reads bronze_orders as a stream, a Delta table being as good a source as Kafka, and counts only placed events, so a payment does not count an order twice. window("event_time", "1 hour") puts each event in the hour it was placed. withWatermark comes before the aggregation, as it must. Append mode writes each hour once, when it is final, so hourly_sales never holds a number that later changes. At the end it prints what Spark reported for each batch: events in, events too late, and the watermark.
python hourly_sales.py prints:
batch 0: 4 events in, 0 too late, watermark 1970-01-01T00:00:00.000Z
batch 1: 0 events in, 0 too late, watermark 2026-10-01T09:20:00.000Z
And python show.py hourly_sales shows an empty table. Batch 0 read the 4 placed orders, starting with no watermark yet. Spark sets the watermark at the start of each batch from what it has seen, and ran batch 1, with no new data, only to apply it: the newest order is 10:20, less 1 hour is 09:20, and the 09:00 hour ends at 10:00, after that. So no hour is final yet, and append mode writes nothing.
Time moves on, and events arrive late
Add 2 batches to BATCHES in make_orders.py: an order at 11:30, then 3 orders together, one of them from 09:50, held up for nearly 2 hours, and one from 10:45:
"3": [
(1005, "C02", "placed", 18.25, "2026-10-01T11:30:00Z"),
],
"4": [
(1006, "C03", "placed", 22.00, "2026-10-01T09:50:00Z"),
(1007, "C02", "placed", 15.00, "2026-10-01T10:45:00Z"),
(1008, "C01", "placed", 12.00, "2026-10-01T12:40:00Z"),
],
Send batch 3, load it, and count:
python make_orders.py 3
python read_orders.py
python hourly_sales.py
batch 2: 1 events in, 0 too late, watermark 2026-10-01T09:20:00.000Z
batch 3: 0 events in, 0 too late, watermark 2026-10-01T10:30:00.000Z
11:30 moved the watermark to 10:30, past the end of the 09:00 hour, and hourly_sales now has its first row: 2026-10-01 09:00:00, 3 orders, 68.50. Now batch 4, with the same 3 commands:
batch 4: 3 events in, 1 too late, watermark 2026-10-01T10:30:00.000Z
batch 5: 0 events in, 0 too late, watermark 2026-10-01T11:40:00.000Z
+-------------------+------+-------+
|hour |orders|revenue|
+-------------------+------+-------+
|2026-10-01 09:00:00|3 |68.50 |
|2026-10-01 10:00:00|2 |55.00 |
+-------------------+------+-------+
1 event was too late: 1006, from 09:50. Its hour had been written already, so it was dropped. 1007, from 10:45, was late too, but the 10:00 hour was still open, since the watermark of 10:30 had not reached 11:00, so it counts: 40.00 plus 15.00. 12:40 then moved the watermark to 11:40, and the 10:00 hour was written. The 11:00 and 12:00 hours are not written yet, and will not be until an order after 13:00 moves the watermark past 12:00. A stream that goes quiet holds back its last hours.
The numbers a stream did not count
bronze_orders still has 1006: the watermark only decides what hourly_sales counts. Count every hour again from bronze, as a nightly batch would. Save this as recount.py:
"""Count every hour again from all of bronze, late events included, as a nightly batch would."""
from pyspark.sql import functions as F
from stream_session import stream_spark
spark = stream_spark("recount")
(
spark.table("bronze_orders")
.where("status = 'placed'")
.groupBy(F.date_trunc("hour", "event_time").alias("hour"))
.agg(F.count("*").alias("orders"), F.sum("amount").alias("revenue"))
.orderBy("hour")
.show()
)
+-------------------+------+-------+
| hour|orders|revenue|
+-------------------+------+-------+
|2026-10-01 09:00:00| 4| 90.50|
|2026-10-01 10:00:00| 2| 55.00|
|2026-10-01 11:00:00| 1| 18.25|
|2026-10-01 12:00:00| 1| 12.00|
+-------------------+------+-------+
The 09:00 hour sold 90.50, not 68.50: the 22.00 of order 1006. Neither number is wrong. The stream's says what was known when the hour was closed; the recount says what is known now. Keep the raw events, and let a batch correct the totals for anything later than the watermark, as the scenario below works through.
ScenarioThe daily sales total that grows after midnightMonday's sales on the dashboard say $412,600. Finance runs the same query on Thursday and gets $426,600. Nobody changed the data, and both numbers are right for…Choosing the delay
The delay is a trade. A short one writes each hour soon and keeps little state, but drops more late events; a long one counts more of them, but holds more hours open in memory and writes each one later. Measure how late your events really are, such as the gap between event time and Kafka's timestamp column, and choose a delay above nearly all of them.
Append mode is not the only choice. Update mode writes an hour's row every time it changes, so a dashboard sees the 10:00 hour at once and sees it change; writing that to a Delta table needs a MERGE on the hour, which part 3 does with foreachBatch. Complete mode rewrites every hour each batch and can never drop state, so it suits only small results. The same watermark also bounds other state: dropDuplicatesWithinWatermark(["order_id", "status"]) removes events sent twice, remembering each key only as long as the watermark allows.
Part 3 asks what happens to these counts when a stream fails halfway through a batch.
Check yourself
2 questions, each in a new situation
1. hourly_sales has written the 09:00 and 10:00 hours, but not 11:00, although the 11:30 order was read long ago. What has to happen for 11:00 to be written?
Not quite. The newest event so far is 12:40, so the watermark is 11:40, and the 11:00 hour ends at 12:00. In append mode a window is written only once the watermark passes its end. A quiet stream holds back its last hours, which is the price of waiting for late events; update mode writes sooner but its rows can still change.
2. Order 1006, placed at 09:50, arrived after the stream had seen 11:30. The watermark was then 10:30. Why was it dropped, while 1007, placed at 10:45, counted?
Not quite. Both were late, but what matters is whether their window was still open. The watermark was 10:30, so the hour ending at 10:00 was closed and the hour ending at 11:00 was not. Spark only promises to count events less late than the delay; this one was 1 hour 40 minutes behind the newest event.
Questions
- What is the difference between event time and processing time in Spark Structured Streaming?
- Event time is when something happened, a column in the data such as when an order was placed. Processing time is when Spark reads it. Grouping by event time with window() counts each event in the hour it happened, however late it arrives, as long as it arrives before the watermark passes; grouping by processing time would count it in the hour it was read.
- Does Spark drop all data that is later than the watermark?
- No. Spark guarantees to count data less late than the watermark's delay, but the guarantee is strict only in that direction: data later than the delay may or may not be counted. In append mode, an event for a window that has already been written is dropped, and the stream's progress reports how many rows were dropped that way.
Sources