Watermarking and Late Data Handling in Spark Structured Streaming

Scenario context

A watermark tells Spark Structured Streaming: "I will accept events up to N minutes late, and then I am done waiting." Spark tracks the maximum event time seen per partition, takes the global minimum across all partitions, subtracts the threshold, and uses that as the watermark. Windows whose end time falls before the watermark are finalized and evicted from the state store. Setting the threshold too short drops rea…

This walkthrough focuses on The Late Payment Problem: When Waiting Forever Is Not an Option.

Before making a design choice, clarify:

• A mobile payments company processes hundreds of thousands of transactions per second.
• Their fraud detection pipeline aggregates payment events by merchant and computes rolling five-minute totals to spot unusual velocity patterns.
• The source is a Kafka topic; the sink is a Delta Lake table consumed by a risk-scoring service.

Concepts in play: spark-data-engineering, event-driven-architecture, core-algorithm-patterns, data-warehousing-lakehouse

Stage 1 — Event Time, Processing Time, and What It Means for a Window to Be "Late"

• Before watermarks make sense, it is essential to understand the three different clocks that exist inside a streaming system — and why only one of them reflects what actually happened in the physical world.
• Event time is the timestamp embedded in the event itself, reflecting when the action occurred.
• For a payment, it is the moment the cardholder tapped their phone.
• This is the timestamp that matters for business logic: a fraud velocity check must use event time, not arrival time, because a genuine burst of spending looks different from a delayed batch of retransmissions.

flowchart TD
    A["New event arrives"] --> B{"Window type?"}
    B -- Tumbling --> C["Assign event to exactly one fixed window"]
    B -- Sliding --> D["Assign event to multiple overlapping windows"]
    B -- Session --> E["Extend existing session or open new session gap"]
    C --> F["One state entry updated"]
    D --> G["Multiple state entries updated - higher memory cost"]
    E --> H["Dynamic window boundary - variable state size"]
    F --> I["Watermark passes window end - finalize and evict"]
    G --> I
    H --> I

Observable outcome: This is the timestamp that matters for business logic: a fraud velocity check must use event time, not arrival time, because a genuine burst of spending looks different from a delayed batch of retransmissions.

Stage 2 — How Spark Tracks the Watermark and Decides When a Window Is Safe to Close

• The watermark is not a static value.
• It advances forward as Spark observes newer event times in the data stream.
• Understanding how that advancement works — and what it controls — is the key to sizing thresholds correctly.
• The watermark formula is simple: When Spark processes each micro-batch, it scans all the event timestamps in that batch, finds the maximum, and updates its internal watermark tracker if this maximum exceeds the previous…

current watermark = max event time seen so far − late threshold

flowchart TD
    A["Event arrives in micro-batch"] --> B["Extract event timestamp"]
    B --> C["Compute max event time in this micro-batch"]
    C --> D["Update watermark = max event time - threshold"]
    D --> E{"Is event timestamp >= watermark?"}
    E -- Yes --> F["Accept event - assign to window"]
    E -- No --> G["Drop event - too late"]
    F --> H["Update partial aggregation in state store"]
    H --> I{"Watermark now exceeds window end?"}
    I -- Yes --> J["Finalize window - emit result - evict state"]
    I -- No --> K["Window stays open in state store"]

Observable outcome: The watermark formula is simple: When Spark processes each micro-batch, it scans all the event timestamps in that batch, finds the maximum, and updates its internal watermark tracker if this maximum exceeds the previous…

Stage 3 — Deep Dive into Watermark Internals and State Store Performance

• The Internals of Watermark Tracking Across Partitions and Micro-Batches The phrase "max event time seen so far" conceals an important distributed-systems subtlety.
• Spark does not have a single global event-time tracker.
• Each executor processes its own partition of the Kafka input stream independently.
• Events in partition 0 may be several minutes ahead of events in partition 3 if one Kafka consumer is lagging.

flowchart TD
    Driver["Driver Node"]
    Planner["Query Planner"]
    Optimizer["Catalyst Optimizer"]
    Executor["Executor Task"]

    Driver -->|"Spark Execution"| Planner
    Planner --> Optimizer
    Optimizer -->|"Optimized Plan"| Executor
    Executor -->|"Result"| Driver

Observable outcome: Events in partition 0 may be several minutes ahead of events in partition 3 if one Kafka consumer is lagging.

Stage 4 — How Events Flow Through Watermark Tracking to Window Finalization

• The following diagram traces the complete path an event takes from Kafka partition arrival through per-partition tracking, global watermark computation, window finalization, and state eviction.
• Each step is a distinct responsibility in the Spark internals.
• Read this diagram from top to bottom: three Kafka partitions deliver events processed by three executors, each tracking its local maximum event time.
• The Driver collects these maxes and takes the global minimum (14:09 — the slowest partition sets the pace).

graph TD
    KafkaP0[Kafka Partition 0 - max event time 14:12]
    KafkaP1[Kafka Partition 1 - max event time 14:09]
    KafkaP2[Kafka Partition 2 - max event time 14:11]
    Executor0[Executor 0 - tracks local max 14:12]
    Executor1[Executor 1 - tracks local max 14:09]
    Executor2[Executor 2 - tracks local max 14:11]
    Driver[Driver - collects per-partition maxes]
    GlobalMin[Global Min = 14:09 with threshold 5 min]
    Watermark[New Watermark = 14:04]
    StateStore[State Store - scan for finalizable windows]
    FinalizeWindow[Finalize windows ending before 14:04]
    EmitResult[Emit window results in append mode]
    EvictState[Evict finalized state entries]
    DropLate[Drop late events with event time before 14:04]

    KafkaP0 --> Executor0
    KafkaP1 --> Executor1
    KafkaP2 --> Executor2
    Executor0 --> Driver
    Executor1 --> Driver
    Executor2 --> Driver
    Driver --> GlobalMin
    GlobalMin --> Watermark
    Watermark --> StateStore
    StateStore --> FinalizeWindow
    FinalizeWindow --> EmitResult
    FinalizeWindow --> EvictState
    Watermark --> DropLate

Observable outcome: The Driver collects these maxes and takes the global minimum (14:09 — the slowest partition sets the pace).

Stage 5 — Where Watermarks Shape Real Production Streaming Pipelines

• Tumbling-window payment aggregations with withWatermark.
• The canonical use case: aggregate payment events by merchant in 5-minute tumbling windows to feed a fraud detection model.
• A 10-minute watermark threshold accommodates typical mobile network retries and brief signal loss.
• Windows are finalized 15 minutes after their start time in steady-state operation (5-minute window duration plus 10-minute threshold): a healthy trade-off between latency and completeness.

Observable outcome: Windows are finalized 15 minutes after their start time in steady-state operation (5-minute window duration plus 10-minute threshold): a healthy trade-off between latency and completeness.

Stage 6 — Choosing Your Watermark Threshold: Latency Requirement Versus Late Data Tolerance

• Use this table as a starting point.
• Measure your actual late-event distribution from Kafka consumer lag metrics before committing to a threshold in production.
• The compliance-grade row is important: when business requirements demand both real-time output and guaranteed completeness, a single watermark threshold cannot satisfy both.
• The standard solution is a Lambda architecture: the streaming pipeline with a conservative threshold provides near-real-time results, and a nightly batch job re-processes the raw events with exact correctness to produce…

Observable outcome: The standard solution is a Lambda architecture: the streaming pipeline with a conservative threshold provides near-real-time results, and a nightly batch job re-processes the raw events with exact correctness to produce…

End-to-end outcome

A mobile payments company processes hundreds of thousands of transactions per second.

Alternatives considered

The source focuses on one primary design. A second architecture is not compared, so no alternative is invented here.

Failure and degraded modes

The Completeness-vs-Latency Tradeoff and Where Output Mode Fits In

• Detection: Every watermark threshold decision is a point on a single spectrum: Short threshold (1–5 minutes): - Watermark advances quickly → windows finalize sooner → state is small → output latency is low.
• User or operator impact: Every watermark threshold decision is a point on a single spectrum: Short threshold (1–5 minutes): - Watermark advances quickly → windows finalize sooner → state is small → output latency is low.