Streaming Joins & Windowing in Spark: Watermarks, Windows & State

Last Updated: July 2026 | 16 min read

Quick Answer: Streaming joins and windowing in Spark Structured Streaming let you correlate two live Kafka streams and aggregate events over event-time windows — but both are stateful, so Spark must buffer data in memory until it's sure no more matching or in-window records will arrive. A watermark (maximum event time seen − late threshold) is what tells Spark when that moment has passed, so it can emit the result and evict old state. Without a watermark, windowed aggregations and stream-stream joins buffer state forever and eventually crash the job with an out-of-memory error.

Every real-time pipeline eventually hits the same wall: you need to join two Kafka streams — say clicks and impressions — or count events per 5-minute window, and the job runs fine for an hour, then the executors start OOMing. The cause is almost never the join logic. It's unbounded state growing because Spark was never told when it's safe to forget old data. This guide covers how event-time windows, watermarks, and stream-stream joins actually work in Spark Structured Streaming, with real PySpark code you can adapt — and, more importantly, how to keep state bounded so your job runs for months instead of hours.

How stateful streaming works: the mental model

Stateful streaming operations keep a "state store" of partial results between micro-batches, and a watermark decides when entries in that store are old enough to finalize and delete. Get this one idea and everything else follows.

In batch processing you have the whole dataset, so a GROUP BY window or a JOIN is easy — every row is present. In streaming, data arrives continuously and out of order. An event stamped 10:04:59 might show up at 10:06 because of a network delay or a Kafka rebalance. So when Spark computes "count per 5-minute window", it can't close the [10:00, 10:05) window the instant the clock hits 10:05 — a late event for that window might still be coming.

The state store holds those still-open windows (or unmatched join rows) in memory, checkpointed to durable storage each batch. The problem: how long do you keep waiting? Wait forever and state grows without bound. Stop too early and you drop valid late data. The watermark is the knob that answers this.

What is a watermark, and how does Spark compute it?

A watermark is a moving timestamp defined as (maximum event time Spark has seen) − (your late-data threshold). Any record with an event time earlier than the current watermark is considered too late and is dropped; any window whose end is behind the watermark is finalized and evicted from state.

You declare it with withWatermark before the stateful operation:

from pyspark.sql import functions as F

events = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "events")
    .option("startingOffsets", "latest")
    .load())

parsed = (events
    .select(F.from_json(F.col("value").cast("string"), schema).alias("d"))
    .select("d.*")
    .withColumn("event_time", F.col("event_time").cast("timestamp")))

windowed = (parsed
    .withWatermark("event_time", "10 minutes")          # tolerate 10 min of lateness
    .groupBy(F.window("event_time", "5 minutes"))
    .agg(F.count("*").alias("events")))

Here Spark tolerates data up to 10 minutes late. If the latest event it has seen is stamped 10:30, the watermark sits at 10:20. The [10:10, 10:15) window is fully behind the watermark, so Spark emits its final count and deletes it from state. A record for 10:12 arriving now (event time 10:12 < watermark 10:20) is silently dropped.

The watermark only ever moves forward. It tracks the max event time seen, so a burst of old data can't drag it backward. This is what guarantees state eventually gets cleaned up even under bursty, out-of-order load.

Choosing the late threshold

The threshold is a latency-vs-completeness trade-off, and it's a business decision, not a technical default:

• Too short (e.g. 1 minute) → lower latency and smaller state, but you drop legitimately late events and undercount.
• Too long (e.g. 6 hours) → you capture nearly all late data, but every window and join row lives in state for 6 hours, ballooning memory.

Measure your real lateness. Pull the event-time-vs-processing-time gap from your source for a day and set the threshold near the 99th percentile. At Telemetrix, our device telemetry is 99% on time within ~4 minutes but has a long tail from devices on flaky networks; we set a 15-minute watermark and accept that the last 0.1% is dropped rather than pay to buffer state for hours.

Event-time windows: tumbling, sliding, and session

Spark supports three window types — tumbling (fixed, non-overlapping), sliding (fixed, overlapping), and session (dynamic, gap-based) — and the right one depends entirely on the question you're answering. All three key off the event-time column and require a watermark to bound state.

Tumbling window

# One row per 5-minute bucket — each event counted once
.groupBy(F.window("event_time", "5 minutes"))

Sliding window

# 10-min window, emitted every 5 min — each event lands in 2 overlapping windows
.groupBy(F.window("event_time", "10 minutes", "5 minutes"))

The second argument is the slide duration. When it's smaller than the window length, windows overlap and each event is aggregated into every window it falls inside — great for smooth rolling metrics, but it multiplies your state footprint.

Session window

# Window stays open until 30 minutes of inactivity for that user
.groupBy(F.col("user_id"), F.session_window("event_time", "30 minutes"))

Session windows (native since Spark 3.2) don't have a fixed size — they expand as long as events for the same key keep arriving within the gap duration, then close. This is the correct tool for "how long was this user active?" that fixed windows can't express.

Stream-stream joins: correlating two live streams

A stream-stream join matches rows from two continuous streams — but because either side can arrive late, Spark must buffer past rows from both streams in state until it's certain no future match is possible. A watermark on both streams plus a time-range condition in the join is what makes that certainty (and state cleanup) possible.

The classic example: join an impressions stream to a clicks stream to attribute clicks to the ad that produced them, where a click can arrive minutes after the impression.

impressions = (spark.readStream.format("kafka")
    .option("subscribe", "impressions").load()
    # ...parse...
    .withColumn("imp_time", F.col("imp_time").cast("timestamp"))
    .withWatermark("imp_time", "1 hour"))

clicks = (spark.readStream.format("kafka")
    .option("subscribe", "clicks").load()
    # ...parse...
    .withColumn("click_time", F.col("click_time").cast("timestamp"))
    .withWatermark("click_time", "30 minutes"))

attributed = impressions.join(
    clicks,
    F.expr("""
        impression_id = click_impression_id AND
        click_time >= imp_time AND
        click_time <= imp_time + interval 30 minutes
    """)
)

Two things make this join bounded:

1. withWatermark on both streams — tells Spark how late each side can be.
2. The time-range condition (click_time BETWEEN imp_time AND imp_time + 30 min) — tells Spark that a click more than 30 minutes after an impression can never match, so once the watermark passes imp_time + 30 min, that impression row is safe to drop from state.

Without the time bound, Spark has no way to know a match is impossible, so it keeps every impression in state forever. This single omission is the most common cause of stream-stream join OOMs.

Watermark requirements by join type

For outer joins the watermark isn't just about memory — it's about correctness. Spark can only emit a (impression, NULL) outer row once it's sure no click will ever match, and the watermark is the only signal that provides that guarantee.

Global watermark gotcha: when two streams have different watermarks, Spark defaults to a single global watermark = the minimum across both streams. The slower stream drags the whole query's eviction back. If one stream lags, state for the other piles up. Set spark.sql.streaming.multipleWatermarkPolicy=max to let the faster stream drive eviction — but only if you're willing to drop more late data from the slow side.

Output modes: which one goes with which operation

Output mode controls what Spark writes each batch, and not every mode works with every stateful operation. Picking the wrong one throws an AnalysisException at query start.

For a windowed aggregation written downstream, append is almost always right: each window is emitted exactly once, after the watermark passes it. If you need the running value every batch (a live dashboard), use update. Reserve complete for tiny result sets — it re-emits everything each batch and does not evict state.

Common mistakes (and how to avoid them)

Most streaming-join and windowing failures come from a short list of mistakes:

1. No watermark before a stateful op. State grows every batch until OOM. Always withWatermark on the event-time column before the groupBy/join/dropDuplicates.
2. Watermark on the wrong column. The watermark must be on the same event-time column the window or join range uses. Watermarking a column the operation ignores does nothing.
3. Stream-stream join with no time-range condition. An equality-only join can never bound state. Add the BETWEEN/interval constraint.
4. Using processing time instead of event time. current_timestamp() windows are trivially "on time" but wrong — late and out-of-order data is mis-bucketed. Window on the event's own timestamp.
5. Late threshold set by guesswork. Measure the real event-time lag distribution; don't pick 1 hour because it "feels safe" — that's an hour of state.
6. Ignoring state skew. One hot join key (a bot user, a null id) can bloat a single partition's state store, the streaming cousin of Spark data skew. Filter nulls and hot keys before the join.
7. Forgetting checkpoint compatibility. Changing a stateful query's watermark or keys can break checkpoint recovery. Use a new checkpoint path for incompatible changes.

See how SolutionGigs can help

Building a real-time pipeline and not sure where the state is leaking, or whether your architecture even needs a stream-stream join versus a simpler batch approach? SolutionGigs connects you with vetted data engineers who've shipped production Spark + Kafka streaming systems. See how SolutionGigs can help →

Frequently Asked Questions

What is a watermark in Spark Structured Streaming?

A watermark is a moving threshold that tells Spark how long to wait for late data before finalizing a window and dropping its state. Spark computes it as the maximum event time seen so far minus your late threshold. Any record older than the current watermark is dropped, and any window whose end is behind the watermark is emitted and evicted. Watermarks are what let windowed aggregations and stream-stream joins run with bounded memory.

Do stream-stream joins require a watermark in Spark?

For inner joins a watermark plus a time-range condition is technically optional but essential in practice — without them Spark buffers both streams forever and eventually OOMs. For outer joins (left, right, full) the watermark and an event-time constraint are mandatory, because Spark needs them to know when a row will never match so it can emit the NULL-padded result and free state.

What is the difference between tumbling, sliding, and session windows?

Tumbling windows are fixed and non-overlapping, so each event lands in exactly one bucket. Sliding windows overlap — a 10-minute window sliding every 5 minutes counts each event in two windows, smoothing trends at the cost of more state. Session windows are dynamic: they stay open while events keep arriving within a gap duration and close after inactivity, which fits user sessions of unpredictable length.

Why does my Spark streaming state keep growing until the job crashes?

Unbounded state growth almost always means a stateful operation is running without a watermark, or the watermark is on a column the operation doesn't use. A windowed aggregation, dropDuplicates, or stream-stream join with no watermark can never decide old state is safe to delete, so it grows every batch until the executor OOMs. Add withWatermark on the event-time column before the operation and key the operation off that same column.

Which output mode should I use with windowed aggregations?

Use append mode when you want each window emitted once, after the watermark passes its end — the right choice for writing to Kafka, files, or Iceberg. Use update mode when you want the running result emitted every batch, which suits dashboards and upserts. Complete mode re-emits the entire result table each batch and should be reserved for small aggregations, never high-cardinality keys.

How does Spark handle watermarks when joining two streams with different delays?

By default Spark uses a single global watermark equal to the minimum event time across both streams, so the slowest stream governs eviction for the whole query. If one stream lags, the other's state lives longer than needed. You can set spark.sql.streaming.multipleWatermarkPolicy to max so the fastest stream drives eviction — but only if you accept dropping more late data from the slow stream.

Can I join a stream to a static table instead of another stream?

Yes. A stream-static join (stream joined to a batch DataFrame, such as a lookup or dimension table) is stateless and needs no watermark, because the static side is fully known. It's often the simpler, cheaper choice when one side is reference data that changes slowly — reach for a stream-stream join only when both sides are genuinely continuous.

Conclusion

Streaming joins and windowing are where most Spark pipelines either scale for months or fall over in hours — and the deciding factor is almost always state discipline, not clever logic. Windows and stream-stream joins are stateful, and Spark can only bound that state if you give it a watermark on the right event-time column and, for joins, a time-range condition that makes future matches provably impossible.

The playbook that keeps production streaming jobs healthy:

• Always withWatermark on the event-time column, before every stateful operation.
• Match the window type to the question — tumbling for per-interval metrics, sliding for rolling trends, session for activity bursts.
• Bound stream-stream joins with a time-range condition, and use mandatory watermarks for any outer join.
• Pick the output mode that fits the sink — append for finalized windows, update for live counters.
• Measure your real lateness and set the threshold from data, not superstition.

Once state is bounded, the rest of tuning — triggers, shuffle partitions, checkpointing — is covered in the Spark Streaming tuning guide. And if your streaming architecture is getting complex enough that you'd rather have an expert review it, get matched with a vetted data engineer on SolutionGigs — it's free to post your project.

Mohammed Yaseen

Founder, SolutionGigs

Mohammed builds Spark Structured Streaming pipelines on AWS EMR at Telemetrix, joining and windowing hundreds of millions of Kafka telemetry events a day into Iceberg on S3. He's debugged the state-store OOMs, global-watermark surprises, and late-data drops this guide warns about — in production, not in theory. LinkedIn →