Watermarks are a contract: validating 150+ PySpark streams

Every courier company has a cutoff time. Parcels that reach the depot after the truck leaves do not get delivered today. Nobody calls the customer to apologize. The parcel just quietly misses the truck.

Streaming systems run on the same idea. A PySpark watermark is the truck's departure time: events that arrive after the bound still exist, but they no longer count. Set it too early and you quietly lose data. Set it too late and the job holds state forever.

I have met this problem twice in different forms over the last year. Once on a production telemetry pipeline, and once in a simulation we co-built at work. The lesson from both was the same: a watermark is not a config value. It is a promise about lateness, state, and what the job will silently drop.

The production version: 150+ streams

The pipeline processes millions of telemetry events per day across more than 150 streams. The watermark was never the hard part. Knowing whether the promise still held was.

There are three separate questions hidden in withWatermark:

df = (
    events
    .withWatermark("event_time", "10 minutes")
    .groupBy(window("event_time", "5 minutes"), "service")
    .count()
)

1. Completeness. How late can an event be before it will not be included in the window?
2. State. How long must the job remember old keys and windows?
3. Latency. How long do we wait before emitting a result?

Changing the watermark changes all three. A tight bound reduces state and latency, but drops more late data. A loose bound keeps more data correct, but grows state and delays output. There is no universally correct value. There is only an explicit one.

So I stopped trusting a watermark because the job did not crash. I trusted it when three checks passed.

Check 1: late-event accounting

Every micro-batch emits counters for input rows and for rows with an event time below the current watermark. Late rows need a number and a destination. A silent drop is not a validation strategy.

metrics = (
    batch
    .select(
        F.count("*").alias("rows"),
        F.sum(
            F.when(F.col("event_time") < F.col("current_watermark"), 1).otherwise(0)
        ).alias("late_rows"),
    )
)

Check 2: window reconciliation

For a bounded event-time range, recompute the same aggregation as a batch job and compare it with the streaming result. A few late-arriving records can make a window differ legitimately, so the comparison must account for the watermark and the observation time. A difference outside the expected lateness window is a bug or an undocumented behavior. Either way, someone should know.

streamed = spark.read.parquet("gold/service_windows")
batched = (
    spark.read.parquet("silver/events")
    .where(F.col("event_time").between(start, end))
    .groupBy(F.window("event_time", "5 minutes"), "service")
    .count()
)

diff = streamed.exceptAll(batched).unionByName(batched.exceptAll(streamed))

Check 3: schema checks

Streaming jobs tend to fail in the least interesting ways: a column becomes required, a timestamp changes format, an upstream producer starts emitting a string where a number used to be. Validate against a contract and quarantine the rejects instead of dropping the whole batch. Count them. Alert when the rate moves.

Make the sink idempotent

Exactly-once is a property of the whole pipeline, not a word you add to a slide. Retries, reprocessing, and manual backfills still happen. Use a deterministic key and an idempotent write pattern, and test the restart path: stop the job, let late data accumulate, restart it, and check the output against batch truth.

The same promise, a different clock: the POC version

Earlier this year my team and I co-built a medallion POC on simulated HVAC telemetry: 81 chillers, roughly 140K readings a day, PySpark and Delta Lake. There the watermark was not minutes. It was dates.

The job generated one day of data per run, and the question was simpler but the same in spirit: which dates have already been processed? The answer lived in a small JSON file:

{
  "latest_date": "2026-09-14",
  "generated_dates": ["2026-09-10", "2026-09-11", "2026-09-12"],
  "run_history": []
}

The next run starts the day after latest_date:

if watermark_date:
    target_date = watermark_date + timedelta(days=1)

And it refuses to regenerate dates that already exist:

already_generated = requested_dates & set(wm.get("generated_dates", []))
if already_generated == requested_dates:
    print("::notice::Duplicate Generation Prevented.")
    sys.exit(0)

Two details made this honest. Every run appended to run_history, so the file was an audit trail and not just a pointer. And if the watermark upload failed after data was synced, the next run did not trust the stale file. It scanned the bucket's raw directories, reconstructed the real list of dates, and repaired the watermark from what was actually there:

if wm is None:
    print("No watermark found. Scanning bucket to discover existing data...")
    bucket_dates = _scan_bucket_for_dates(bucket_id)

A watermark that can repair itself from ground truth is far more trustworthy than one you hope was written correctly.

Both watermarks are the same contract

The streaming version decides which events count inside a moving time window. The batch version decides which days have been done. Both are promises about time, and both fail quietly when nobody checks them.

The fix in both places was identical:

• Write the expectation down, per stream or per run.
• Emit metrics for everything that fell outside the bound.
• Keep an audit trail a human can inspect.
• Build a reconciliation path that compares against ground truth.

Then the watermark stops being a mysterious parameter that happens to make the dashboard look right. It becomes one line of code implementing a documented decision.

The useful shift is from "the stream is running" to "the stream is keeping its promise."