# 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.

- Page: https://tecyfy.com/data-engineering/course/streaming/event-time
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 9: Streaming, Event time, windows and watermarks
- Parts of this module: Kafka into a Delta table (https://tecyfy.com/data-engineering/course/streaming); Event time, windows and watermarks (https://tecyfy.com/data-engineering/course/streaming/event-time); Exactly-once and foreachBatch (https://tecyfy.com/data-engineering/course/streaming/exactly-once); Streaming on Databricks (https://tecyfy.com/data-engineering/course/streaming/on-databricks); Build it: orders from Kafka to an hourly table (https://tecyfy.com/data-engineering/course/streaming/build-it)

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

[Part 1](/data-engineering/course/streaming) 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:

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/streaming/event-time)

## Orders per hour

Save this as `hourly_sales.py`:

```python
"""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:

```text
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:

```python
    "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:

```bash
python make_orders.py 3
python read_orders.py
python hourly_sales.py
```

```text
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:

```text
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
```

```text
+-------------------+------+-------+
|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`:

```python
"""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()
)
```

```text
+-------------------+------+-------+
|               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.

- Scenario: [The daily sales total that grows after midnight](https://tecyfy.com/practice/scenarios/the-sales-total-that-grows-after-midnight). Monday'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](/data-engineering/course/streaming/exactly-once) asks what happens to these counts when a stream fails halfway through a batch.

## Check yourself

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?

   A. Nothing; it is a bug
   B. The stream has to run with a processing-time trigger
   C. An event placed after 13:00 has to arrive, so the watermark, 1 hour behind the newest event, passes 12:00, the end of the 11:00 hour
   D. Someone has to run recount.py

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?

   A. 1006's hour, 09:00 to 10:00, ended before the watermark, so it was final and written; 1007's hour ends at 11:00, after the watermark, so it was still open
   B. 1006 has a bad amount
   C. Spark drops every event that arrives out of order
   D. 1007 arrived first

### Answers

1. C. 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. A. 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

- [Structured Streaming Programming Guide: APIs on DataFrames and Datasets](https://spark.apache.org/docs/latest/streaming/apis-on-dataframes-and-datasets.html)
- [Apply watermarks to control data processing thresholds](https://docs.databricks.com/aws/en/structured-streaming/watermarks)