Streaming Aggregations with Watermarking

Your streaming job counts clicks per user, but late events keep arriving after you've already written results — so your numbers are wrong, and you can't tell when it's safe to update. Without a strategy for handling late data, you either wait forever (and your dashboard is always behind) or you close the window too early and silently drop events. In Databricks, write streaming aggregations with watermarking solves this by telling the engine how long to wait for late data before finalizing state. By the end of this lesson, you'll understand the core concept, apply it in a hands-on exercise, and know how to connect this skill to your next step in the Databricks learning path.

The problem this lesson solves

Streaming data doesn't respect your schedule. A user clicks a button, but their event might land in Kafka seconds late — or minutes late due to network retries or client-side buffering. If you're computing a streaming aggregation like a count or sum over a tumbling window, you face a dilemma:

• If you emit results as soon as a window closes, you'll miss events that arrive just after that boundary.
• If you wait for every possible late event, you'll hold state forever and your results become stale — and your memory usage explodes.

Without a watermark, your pipeline either:

• Produces inaccurate results that silently change when late data finally arrives, or
• Blocks progress, delaying downstream consumers who need fresh results.

For example, imagine you're counting page views per 5-minute window to trigger a real-time promotion. An event that arrives 30 seconds late would be dropped if you closed the window immediately — costing you a sale. But if you wait 30 minutes for every event, your promotion logic never runs on time.

Watermarking gives you a controlled compromise: you define how late you're willing to wait, and the engine handles the rest — finalizing old windows, cleaning up state, and still allowing most late events to be included.

Why now? In any production streaming pipeline, late data is not an edge case — it's the norm. Master watermarking before you build anything more complex like stateful joins or sessionization.

Core concept / mental model

Think of a watermark as a clock that marks the point on the event-time timeline up to which the engine considers all data received. Anything earlier than the watermark is considered "too late" and is ignored. Events between the watermark and the current processing time are still welcome.

In Databricks (using Structured Streaming), you define a watermark with the withWatermark method on a streaming DataFrame. It takes a column that represents event time (e.g., the timestamp when the event actually happened) and a delay threshold (e.g., 10 minutes). The engine then:

1. Tracks the maximum event time seen so far.
2. Subtracts the delay threshold to get the current watermark.
3. Keeps partial results for windows that haven't passed the watermark.
4. Finalizes and outputs windows once the watermark moves past their end time.

Here's a mental picture:

Event time timeline
------------------>
        maxEventTime  = 12:00
        watermark     = 11:50   (12:00 - 10 min)

        Window [11:45, 11:50)   -> finalized (end <= watermark)
        Window [11:50, 11:55)   -> still open

A tumbling window is a fixed-size, non-overlapping time bucket (e.g., 5 minutes). The engine groups events by window, aggregates them, and emits a new row per window when the watermark moves past the window end.

Key mental model: The watermark isn't a wall that cuts off all late events — it's a moving boundary that ensures you are never more than delay behind the latest event time.

How it works step by step

Follow these steps to add watermarking to a streaming aggregation:

1. Start with a streaming DataFrame — from a source (e.g., readStream on Auto Loader, Kafka, or a Delta table).
2. Ensure you have an event-time column — if your source doesn't provide one, you may need to parse a timestamp field (e.g., from JSON) into a TimestampType column.
3. Call withWatermark on the DataFrame, specifying the event-time column and the allowed lateness (e.g., "10 minutes").
4. Apply a windowed aggregation using groupBy(window(event_time_col, "5 minutes")) and then an aggregate function like count or sum.
5. Write the result to a sink using writeStream (e.g., outputMode("append") or "update") and start the query.

The engine automatically:

• Tracks the maximum event time across all partitions.
• Computes the watermark as max(eventTime) - delay.
• Outputs final results for windows whose end time is <= the watermark (in append mode).
• Drops state for those windows, freeing memory.

Important: You must use the same event-time column in both withWatermark and the window function for watermarking to work. If you use a different column, the watermark won't be applied.

Hands-on walkthrough

Let's simulate a streaming source using rate or a simple socket. For local testing, you can create a tiny Kafka-like flow with readStream.format("rate"), but we'll use a more realistic example with a file source.

Example 1: Basic watermark with tumbling window

Assume you have JSON files landing in /tmp/events/, each containing user_id, event_time, and event_type. Set up a stream with a 10-minute watermark and a 5-minute tumbling window.

from pyspark.sql.functions import window, count
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

schema = StructType([
    StructField("user_id", StringType(), True),
    StructField("event_time", TimestampType(), True),
    StructField("event_type", StringType(), True)
])

# Streaming read from a directory (Auto Loader would be similar)
df = spark.readStream \
    .schema(schema) \
    .json("/tmp/events")

# Apply watermark and windowed aggregation
counts = df \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(window("event_time", "5 minutes"), "user_id") \
    .count()

query = counts \
    .writeStream \
    .outputMode("append") \
    .format("memory") \
    .queryName("clicks_per_user") \
    .start()

query.awaitTermination(timeout=60)

When you run this in a notebook, you'll see results like this in the clicks_per_user table:

Only windows whose end_time is before the watermark appear — that's the guarantee of append mode.

Example 2: Using update mode for stateful results

Some use cases (like a running total) need update mode so you see latest aggregates even for unfinished windows.

# Same source, but use update mode to see partial results
query = counts \
    .writeStream \
    .outputMode("update") \
    .format("console") \
    .start()

query.awaitTermination(timeout=60)

In update mode, you'll see rows for every window, and they get updated as events arrive — but you still can't rely on them being final until the watermark passes the window end.

Pro tip: Use append mode for analytics that need final numbers (like a daily report) and update mode for live dashboards where partial numbers are acceptable.

Example 3: Watermark with event-time parsing

If your source stores event time as a string, parse it to a timestamp before applying the watermark.

from pyspark.sql.functions import to_timestamp, window, sum

# Assume 'ts' column is a string like "2024-01-01 10:00:00"
df = df.withColumn("event_time", to_timestamp("ts"))

aggregated = df \
    .withWatermark("event_time", "5 minutes") \
    .groupBy(window("event_time", "10 minutes")) \
    .sum("amount")

Compare options / when to choose what

Watermarking changes how the engine handles state and output. Here's how it compares to alternatives:

When to choose what

• Use watermark for any production aggregation where event-time accuracy matters — that's 99% of cases.
• Use dropDuplicates when you're deduplicating events (e.g., click IDs) over a time window.
• Avoid processing-time windows unless your data truly has no event-time column and you accept approximation.

Troubleshooting & edge cases

Pro tip: Use spark.conf.set("spark.sql.streaming.schemaInference", "true") for quick tests, but always define an explicit schema in production to avoid performance hits.

What you learned & what's next

You now understand the core pain of late data in streaming aggregations, the mental model of a watermark, and how to write a streaming aggregation with watermarking in Databricks. You can complete the hands-on exercise above — run it in a notebook with mock data and observe how output changes as you adjust the watermark delay.

Recall your learning objectives: you can explain the core idea behind write streaming aggregations with watermarking, and you've completed a practical exercise that demonstrates it. Next in the Databricks learning path, you'll build on this by exploring stateful streaming operations like joins with watermarking or sessionization — where the same principles extend to more complex pipelines.

Continue to the next lesson in the track to keep your momentum!