Why Your PySpark Watermark Stopped Working — and Crashed Your Job

A single misplaced .groupBy() can silently kill your watermark, balloon state memory, and take down your entire streaming pipeline. Here's exactly what happened — and how to fix it.

🔥 Imagine This Nightmare

Your PySpark Structured Streaming job has been humming beautifully in production for weeks.

Logs aggregate perfectly. Late events are handled gracefully. Memory usage is flat as a pancake. Life is good.

Then a code review comment drops: “Can you clean up the timestamp parsing — it looks a bit messy.”

You make a quick refactor. Move a couple of lines around. PR approved. Deploy. You walk away for lunch feeling productive.

You come back to find the job consuming 40 GB of memory — and climbing. Your on-call phone buzzes. The job just OOM-crashed. Your manager is asking questions.

You changed nothing about the logic. You just moved some lines around. Yet your watermark — the guardian that prevents infinite state accumulation — is completely dead.

This is one of the most insidious bugs in PySpark Structured Streaming. It bites experienced engineers constantly. And the worst part? Spark gives you zero warnings. The job runs perfectly — it just silently eats your memory until it dies.

Let’s dissect exactly what went wrong, why it happens, and how to never let it happen again.

🎯 The Problem Statement

Before we dive in, let’s establish what was supposed to be happening.

You had a PySpark Structured Streaming job doing three things:

• Reading log events from Kafka
• Aggregating them over a 10-minute tumbling event-time window
• Using a 20-minute watermark to handle late-arriving events

That watermark is the critical piece. Without it, Spark has no idea when it is safe to throw away old window state. So it keeps everything — every window, every micro-batch — in memory. Forever. Until the machine runs out of heap and crashes.

The watermark was your safety valve. And your refactor accidentally broke it.

Why does this matter beyond this one job? Because this pattern — log aggregation, clickstream processing, fraud detection, IoT sensor data — is everywhere in modern data engineering. Understanding this bug deeply is what separates a junior Spark developer from a senior one. It also happens to be a favorite interview question at Databricks, Uber, LinkedIn, and Netflix.

💡 Deep Dive: How Watermarking Actually Works

Let’s build the mental model from the ground up.

The Event-Time Clock

Spark Structured Streaming doesn’t use your wall clock (called processing time) to manage windows. It uses event time — the timestamp embedded inside your actual data. This is important because events often arrive late. A log generated at 12:00 PM might reach your Kafka topic at 12:22 PM due to network delays or device buffering.

Event-time windows handle this correctly. But they create a problem: when is a window “done”? If a window covers 12:00–12:10 PM, Spark can’t know it’s finished until it’s certain no more 12:00–12:10 events are coming. Without a signal, it must keep that window’s state in memory indefinitely.

That signal is the watermark.

What the Watermark Actually Computes

Spark tracks the maximum event-time it has seen across all partitions in every micro-batch. Call this the “high-water mark.” The watermark threshold is then:

Watermark = max(event_time seen so far) − watermark_delay

So with your 20-minute delay:

Latest event seen  →  12:55 PM
Watermark          →  12:55 - 0:20 = 12:35 PM

Any window ending before 12:35 PM → state is finalized and DROPPED ✓

Once the watermark advances past a window’s end time, Spark emits the final result for that window and evicts its state from memory. That is your memory cleanup. That is what was dying in your job.

The Role of withWatermark() in the Logical Plan

When you write a PySpark streaming job, you are building a logical plan — a tree of operations. When Spark compiles that tree into a physical execution plan, it walks from top to bottom looking for an EventTimeWatermark node. For state eviction to work, that node must appear in the plan before the stateful aggregation node.

Think of it like a badge system at a concert. The watermark is the badge. The aggregation is the security gate. The badge must be issued before you reach the gate. If you show up at the gate and then try to get a badge issued on the other side of it, the gate has already let you through without checking — and it will never check again.

🧪 The Bug, Step by Step

The Original Working Code

Here is what your job looked like before the refactor:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_timestamp, window

spark = SparkSession.builder.getOrCreate()
raw = spark.readStream.format("kafka") \
    .option("subscribe", "app-logs") \
    .load()
df = raw \
    .selectExpr("CAST(value AS STRING) AS raw_json") \
    .select(
        col("raw_json.service").alias("service_name"),
        to_timestamp(col("raw_json.ts"), "yyyy-MM-dd HH:mm:ss").alias("event_time")
    ) \
    .withWatermark("event_time", "20 minutes") \   # ① watermark FIRST
    .groupBy(                                        # ② then aggregate
        window("event_time", "10 minutes"),
        "service_name"
    ) \
    .count()

The order here is intentional and critical:

1. Parse the timestamp into a proper TimestampType column named event_time
2. Apply the watermark on that clean, typed column
3. Group by the tumbling window — the aggregation engine sees the watermark and links them together

State is evicted. Memory stays bounded. Everything works.

The Refactored (Broken) Code

After the refactor, it looked something like this:

from pyspark.sql.functions import col, regexp_replace, to_timestamp, window

df_broken = raw \
    .selectExpr("CAST(value AS STRING) AS raw_json") \
    .select(
        col("raw_json.service").alias("service_name"),
        col("raw_json.ts").alias("raw_ts")   # still a STRING!
    ) \
    .groupBy(                                 # ❌ grouped BEFORE watermark
        window(
            to_timestamp(regexp_replace(col("raw_ts"), "T", " ")),
            "10 minutes"
        ),
        "service_name"
    ) \
    .count() \
    .withWatermark("event_time", "20 minutes")   # ❌ too late - aggregation already built

Let’s walk through exactly what went wrong:

Step 1: The timestamp is kept as a raw string raw_ts instead of being parsed into a proper TimestampType column before the aggregation.

Step 2: .groupBy() is called first. Spark builds the stateful aggregation node at this point in the plan — and at this point, there is no watermark definition anywhere in the upstream plan. The aggregation node is constructed without any event-time watermark association.

Step 3: The timestamp transformation (to_timestamp(regexp_replace(...))) happens inline inside the groupBy() call as a native string expression. It never exists as a named, standalone, watermarkable column before the aggregation.

Step 4: .withWatermark() is placed after .count(). It is now operating on the output DataFrame of the aggregation — not on the input. Watermarking the output of an aggregation cannot retroactively configure how that aggregation manages state. The damage is already done.

Result: Spark executes happily. No error. No warning. Just state accumulating in the RocksDB state store, micro-batch after micro-batch, with nothing ever being evicted — until the executor runs out of heap.

How to Confirm the Watermark Is Dead

After your next deployment, open the Spark UI, go to the Streaming tab, and look at the State Operators section. If your watermark is healthy, you will see numRowsDroppedByWatermark increasing over time as old windows expire. If it sits at zero permanently, your watermark is broken.

You can also check programmatically:

import time
time.sleep(60)
print(query.lastProgress["stateOperators"])

# Healthy output:
# {'numRowsDroppedByWatermark': 1523, 'numOutputRows': 847, ...}
# Broken output:
# {'numRowsDroppedByWatermark': 0, 'numOutputRows': 0, ...}

And to inspect the logical plan directly:

df.explain(extended=True)
# Look for EventTimeWatermark BEFORE the Aggregate node.
# If Aggregate appears first, your watermark is not wired in.

⚠️ Common Mistakes and Misconceptions

Mistake 1: Treating watermark position as a style choice. Many engineers think .withWatermark() is just a configuration annotation that can go anywhere. It cannot. Spark's planner is sequential — the watermark node must appear upstream of the stateful aggregation in the DAG.

Mistake 2: Applying watermark after .count() or .agg(). Watermarking the output of an aggregation does nothing for state eviction. You are labeling the wrong layer of the plan. The label needs to be on the input to the aggregation.

Mistake 3: Doing timestamp parsing inline inside groupBy(). When you write to_timestamp(regexp_replace(col("ts"), "T", " ")) inside the groupBy() call, that expression is evaluated as part of the aggregation node itself. There is no named column with a watermark label in the upstream plan for Spark to reference.

Mistake 4: Assuming StringType columns can be watermarked. withWatermark() requires a TimestampType column. Calling it on a string column either raises an error (in newer Spark versions) or silently does nothing. Always cast to timestamp in a dedicated step before watermarking.

Mistake 5: Trusting that Spark will warn you. It won’t. This is a silent failure. The job runs. It just slowly kills itself. The only signal is a memory graph going up forever.

🚀 Pro Tips and Best Practices

Follow the golden order every single time. Parse timestamp → apply watermark → aggregate. Treat this as an unbreakable rule on your team, not a guideline.

Never do timestamp transformations inside groupBy(). Always parse your timestamp in a dedicated .select() or .withColumn() step. It is cleaner, unit-testable, and safe. Inline transformations inside groupBy are a footgun for exactly this reason.

Verify with explain() before deploying. Make it a habit to call df.explain(extended=True) on your streaming DataFrame before pushing to production. Confirm that EventTimeWatermark appears before the Aggregate node. This takes 10 seconds and can save you a 3am incident.

Add a CI test that injects late data. In your staging environment, artificially send some events with timestamps older than your watermark delay. After N micro-batches, assert that numRowsDroppedByWatermark > 0. This programmatically verifies that your watermark is alive with every deployment.

Know your output mode. Watermark-based state eviction only functions in Append output mode. In Complete mode, Spark must maintain all state forever by design — watermarks don't evict anything there. If you need aggregation with state cleanup, use Append mode or Update mode (but Update mode does not support windowed aggregations in all Spark versions — verify your version).

Set your watermark delay conservatively. Your watermark delay should be longer than the longest realistic late-arrival time for your data. If you set it too tight, you will drop legitimate late events. A common pattern is to measure the 99th percentile of your actual event-arrival delay in staging and add a safety margin.

🔄 Real-World Use Cases

Log aggregation pipelines — This is your exact scenario. Every observability platform — Datadog, Splunk, Elastic — uses windowed aggregations over event time to compute error rates, latency percentiles, and throughput metrics. Broken watermarks here cause OOM crashes during traffic spikes, exactly when you need the pipeline most.

Fraud detection at financial institutions — Real-time fraud scoring systems aggregate transaction counts and amounts per user over sliding windows. A broken watermark keeps state objects for millions of users in memory indefinitely. The result is severe GC pressure, latency spikes on scoring, and false negatives — right when low latency matters most.

IoT sensor data processing — Factory sensors, smart meters, and connected devices often transmit data through unreliable networks. Events arrive late by minutes or hours. Watermarks let you accept those late readings while safely discarding the associated state once the window is closed. Without watermarks, a fleet of 50,000 sensors can OOM a Spark cluster within hours.

Clickstream analytics — E-commerce companies track user session behavior over event-time windows to compute real-time engagement metrics. The same watermark pattern applies. Netflix, LinkedIn, and Airbnb all have published engineering blog posts about exactly this class of problem in their own streaming infrastructure.

In interviews, when asked about this scenario, the ideal answer demonstrates three things: you understand Spark’s logical plan construction, you know how to diagnose the issue in a running job, and you know the fix and how to prevent it from happening again. That combination signals senior-level Spark knowledge.

📌 Quick Recap

• Watermark formula: max(event_time seen) − delay. Controls when old window state is evicted from memory.
• Order is not optional: parse timestamp → withWatermark() → groupBy(window(...)). This sequence must be followed.
• Placing .groupBy() before .withWatermark() means the aggregation engine builds its state store without a watermark — state is never evicted.
• Doing timestamp transforms inline inside groupBy() prevents Spark from seeing a named, watermarkable column before aggregation.
• The bug produces no error and no warning. The only signal is unbounded memory growth leading to OOM crash.
• Diagnose with df.explain(extended=True) — look for EventTimeWatermark before Aggregate in the plan.
• Verify at runtime with query.lastProgress["stateOperators"]["numRowsDroppedByWatermark"] — if it's zero, the watermark is dead.
• Watermark eviction only works in Append output mode.
• Fix: always parse your timestamp in a separate .select() step before calling .withWatermark().

Share this article with someone preparing for a data engineering interview — this exact scenario comes up at Databricks, Uber, LinkedIn, and Netflix.

The best way to learn distributed systems is to understand how they break. Keep building. 🚀

🚀 Level Up Your Career — Don’t Wait, Start NOW!

If you’re serious about growing in tech and staying ahead of the curve, this is your moment. No shortcuts — just real skills that actually make a difference.

🌐 Let’s Connect & Grow Together

Follow me for practical insights, real-world learning, and career tips:

🐦 Twitter: https://x.com/SriwWorld 📺 YouTube: https://www.youtube.com/@sriwworldofcoding?sub_confirmation=1 ✍️ Medium: https://medium.com/@sriwworldofcoding 🧵 Threads: https://www.threads.com/@sriwworldofcoding 📸 Instagram: https://www.instagram.com/sriwworldofcoding/ 📘 Facebook: https://www.facebook.com/profile.php?id=61576419014220 🌌 Bluesky: https://bsky.app/profile/sriwworldofcoding.bsky.social

🎯 Want Real Skills? Start With These Hands-On Courses

⚙️ Apache Airflow Bootcamp (Workflow Automation)

👉 https://www.udemy.com/course/apache-airflow-bootcamp-hands-on-workflow-automation/ 💡 Go from beginner to advanced — master DAGs, scheduling, operators, sensors, and build real production workflows.

🔥 PySpark for Data Engineers (Architecture + Interviews)

👉 https://www.udemy.com/course/pyspark-for-data-engineers-architecture-interviews/ 💡 Deep dive into Spark architecture, optimization, and performance tuning — plus crack interviews with confidence.

☁️ Crack Azure Data Engineer Interviews: The Ultimate Q&A Guide

👉 https://www.udemy.com/course/crack-azure-data-engineer-interviews-the-ultimate-qa-guide/ 💡 Get interview-ready with real-world questions on ADF, Synapse, Databricks, Event Hubs, Data Lake, Azure Functions & more.

💥 The difference between where you are and where you want to be? ACTION. Start learning today — your future self will thank you.