Handling Late Data with Allowed Lateness

In this lesson, you will learn how to maintain high-accuracy metrics by configuring your stream processing engine to accept and process events that arrive after the initial watermark has passed.

The Challenge of Late Data

In previous lessons, we defined Watermarks as the threshold for how long the system waits for late-arriving events. Once a watermark progresses past a certain point, the engine assumes no further data for that time window will arrive and clears the state. However, in real-world distributed systems, network partitions or upstream source delays often cause critical events to arrive after that deadline. If we strictly discard this data, our business metrics—like total hourly revenue—will be perpetually understated.

Defining Allowed Lateness

Allowed Lateness acts as a secondary buffer. While the watermark defines when a window is "closed" for output purposes, allowed lateness dictates how long the system keeps the state in memory after that closing time. If a record arrives within the allowed lateness window, the engine will update the previously computed result, potentially triggering a re-calculation or an updated output for that specific time window.

Consider a 10-minute window with a 5-minute watermark and a 10-minute allowed lateness:

1. The window closes at 10:10 AM based on the watermark.
2. The engine emits the result.
3. An event with a timestamp of 10:04 AM arrives at 10:12 AM.
4. Because this falls within the 10-minute allowed lateness (10:10 + 10 = 10:20), the engine re-processes the state and emits an updated aggregate, ensuring your metrics remain accurate despite the delay.

Implementation Concept

When performing stateful aggregations (like groupBy with a window), you apply the lateness parameter to tell the stream that, even though the watermark has passed, the "state" is still valid for a specified duration:

# Conceptualizing the application of allowed lateness
streaming_df \
    .withWatermark("event_time", "5 minutes") \
    .groupBy(window("event_time", "10 minutes")) \
    .count() \
    .option("allowedLateness", "10 minutes") 

By balancing your watermark threshold and your allowed lateness, you can effectively manage the trade-off between latency (how fast you see results) and completeness (how accurate your results are).

Key Takeaway: Allowed Lateness extends the life of your stateful windows, allowing the system to retroactively correct metrics when data arrives slightly past the watermark deadline.