Rana Nasri Ghazzi

Windowing & Watermarking in Databricks Streaming

What Is Windowing?

In a continuous stream, data never stops arriving. Windowing is the mechanism that groups streaming events into finite, time-bounded buckets so you can compute meaningful aggregations — counts, averages, sums — over a defined period rather than over all data since the beginning of time.

Spark Structured Streaming supports three window types.

Tumbling Windows are fixed-size, non-overlapping buckets. Every event belongs to exactly one window — the simplest and most memory-efficient option.

df.groupBy(window("event_time", "10 minutes")).count()
# [0:00–0:10] [0:10–0:20] [0:20–0:30] ...

Sliding Windows are fixed-size but advance at a shorter interval, creating overlap. An event can belong to multiple windows simultaneously, giving you a rolling view of the data.

df.groupBy(window("event_time", "10 minutes", "5 minutes")).count()
# [0:00–0:10] [0:05–0:15] [0:10–0:20] ...

Session Windows are dynamic, defined by inactivity gaps rather than clock time. A window opens when an event arrives and closes after a period of silence. Ideal for user-session analytics.

df.groupBy(session_window("event_time", "5 minutes"), "userId").count()
# Window closes after 5 min of inactivity per user

Watermarking: Taming Late Data

The hard problem with event-time processing: events don’t arrive in order. A sensor reading timestamped at 10:00 might reach Kafka at 10:12 due to network lag, buffering, or a brief device outage. Without watermarking, Spark holds state open for every window forever — just in case something arrives late. That means unbounded memory growth and eventual crashes.

A watermark is a promise: “I will wait up to X minutes for late data. After that, I discard it and close the window.”

df \
  .withWatermark("event_time", "10 minutes") \
  .groupBy(window("event_time", "5 minutes")) \
  .count()

How the watermark moves. It tracks the maximum event timestamp Spark has seen, then subtracts your threshold. It only ever moves forward — never backward.

Max event_time seen:  10:30:00
Watermark threshold:  − 10 minutes
────────────────────────────────
Current watermark:    10:20:00

→ Events with event_time < 10:20 are dropped
→ Windows ending before 10:20 are finalised, state is freed

Output modes control when results are emitted downstream:

• append — emits only finalised (past watermark) windows. Each result written once. Requires watermark.
• update — emits any window that changed this micro-batch, including partial results.
• complete — emits all windows every batch. State grows unboundedly without a watermark.

append + watermark is the production-safe default. It bounds state size, guarantees each result is written exactly once, and prevents OOM on long-running streams.

Sizing Your Watermark in Production

Your watermark threshold is a dial with a direct tradeoff:

The practical approach: measure your actual lag distribution first, then set the watermark to 2–3× your P95 lag.

Why 2–3×? Lag is not constant — it has a distribution. If your median lag is 2 minutes but your P99 is 6 minutes, a watermark of 2 minutes drops roughly 1% of events. A watermark of 6 minutes captures almost everything but adds 6 minutes of output latency. The 2–3× multiplier gives you headroom for lag spikes without blindly targeting an extreme outlier.

Step 1 — Measure lag per event.

To compute lag you need to subtract two timestamps. You can’t do this directly in Spark — the result type is ambiguous. Instead, convert both to integers using unix_timestamp(), which returns seconds since the Unix epoch (1 Jan 1970). Subtracting two integers gives you a clean number of seconds.

event_time = 10:20:00  →  *1705315200
now        = 10:30:45  →           *1705315845
lag        = *1705315845 −- *1705315200 = 645 seconds (≈ 10.75 minutes)

df = df.withColumn(
    "lag_seconds",
    unix_timestamp(current_timestamp()) - unix_timestamp(col("event_time"))
)

Step 2 — Find your distribution.

df.agg(
    avg("lag_seconds").alias("avg_lag"),
    percentile_approx("lag_seconds", 0.95).alias("p95_lag"),
    percentile_approx("lag_seconds", 0.99).alias("p99_lag"),
    max("lag_seconds").alias("max_lag")
).show()

Step 3 — Set and monitor.

# If P95 lag = 3 minutes → set watermark to 6–9 minutes
df.withWatermark("event_time", "7 minutes")

Start wide, then tighten. Run with a conservatively large watermark for 24–48 hours covering peak traffic, collect real lag data, then recalculate. It’s far harder to recover from silently dropped data than to accept a bit of extra output latency.

Monitoring Dropped Events

Once your stream is running, you need to verify your watermark isn’t silently discarding too much data. The key metric is numRowsDroppedByWatermark, which lives inside query.lastProgress — a Python dict updated after each micro-batch.

import json
print(json.dumps(query.lastProgress, indent=2))

The relevant structure in the output:

{
  "batchId": 42,
  "numInputRows": 50000,
  "eventTime": {
    "max": "2024-01-15T10:30:05Z",
    "watermark": "2024-01-15T10:20:05Z"
  },
  "stateOperators": [{
    "numRowsTotal": 50000,
    "numRowsUpdated": 4821,
    "numRowsDroppedByWatermark": 142
  }]
}

To access it directly:

dropped = query.lastProgress["stateOperators"][0]["numRowsDroppedByWatermark"]

Interpreting the number:

• Zero consistently → watermark may be unnecessarily large; consider reducing for lower latency
• Less than 0.1% of input rows → healthy, catching only extreme outliers
• Growing steadily → watermark too small for your lag spikes; increase it
• Sudden spike → upstream outage caused a burst of late data; investigate Kafka lag

Important caveat: numRowsDroppedByWatermark counts rows dropped from state — Spark received them but the target window was already closed. It does not count events so late they never matched a window at all. True data loss may be slightly higher. Treat it as a floor, not a ceiling.

For continuous monitoring, attach a StreamingQueryListener:

from pyspark.sql.streaming import StreamingQueryListener

class WatermarkMonitor(StreamingQueryListener):
    def onQueryProgress(self, event):
        for op in event.progress.stateOperators:
            if op.numRowsDroppedByWatermark > 0:
                pct = op.numRowsDroppedByWatermark / op.numRowsTotal * 100
                print(f"⚠ Dropped: {op.numRowsDroppedByWatermark} rows ({pct:.2f}%)")

spark.streams.addListener(WatermarkMonitor())

Saving Late Rows for Later

When late data matters for correctness, you don’t have to accept the loss. Split your stream before the watermark — route on-time events through the normal aggregation pipeline, and capture late events into a separate Delta table for reprocessing.

Pattern A — Quarantine + Merge (best when late arrivals are frequent and near-realtime correction is needed):

on_time = df.filter(col("lag_seconds") <= 600)
late    = df.filter(col("lag_seconds") >  600)

# Main pipeline
on_time \
  .withWatermark("event_time", "10 minutes") \
  .groupBy(window("event_time", "5 minutes")).count() \
  .writeStream.outputMode("append").format("delta").table("metrics_main")

# Quarantine late rows as-is
late \
  .writeStream.outputMode("append").format("delta").table("metrics_late_arrivals")

A scheduled batch job then re-aggregates and merges them back in:

late_agg = spark.table("metrics_late_arrivals") \
  .groupBy(window("event_time", "5 minutes")).count() \
  .withColumnRenamed("count", "late_count")

DeltaTable.forName(spark, "metrics_main") \
  .alias("m").merge(late_agg.alias("l"), "m.window = l.window") \
  .whenMatchedUpdate(set={"count": col("m.count") + col("l.late_count")}) \
  .whenNotMatchedInsertAll() \
  .execute()

Pattern B — Batch Replay (simpler; use when late data is rare and eventual consistency is acceptable):

spark.read.table("raw_events") \
  .filter("event_time >= now() - INTERVAL 24 HOURS") \
  .withWatermark("event_time", "4 hours") \
  .groupBy(window("event_time", "5 minutes")).count() \
  .write.format("delta").mode("overwrite") \
  .option("replaceWhere", "window >= now() - INTERVAL 24 HOURS") \
  .saveAsTable("metrics_main")

replaceWhere only overwrites the affected time partitions — not the entire table.

TL;DR

• Windowing groups stream events into time buckets. Use tumbling for simple aggregations, sliding for rolling metrics, session for user activity.
• Watermarking tells Spark how long to wait for late events before closing a window and freeing state. Without it, state grows forever.
• Set your watermark to 2–3× your P95 lag, measured from real traffic — not a guess.
• Monitor query.lastProgress[“stateOperators”][0][“numRowsDroppedByWatermark”] to see how many late events are being discarded each batch.
• To recover late rows: split the stream before aggregation, quarantine late events in a Delta table, and merge them back on a schedule.
• Always store raw events. Streams are your fast path. A raw Delta table is your source of truth for reprocessing.