How to Handle Late-Arriving Data and Watermarking in Streaming Pipelines

Watermarks are monotonically increasing timestamps that tell the streaming engine when it can safely close event-time windows and emit aggregates, while allowed lateness configurations determine how long late records can still update results.

In event-driven streaming architectures, data records often arrive out of order due to network latency or processing delays. The DataTalksClub/data-engineering-zoomcamp repository demonstrates production-ready patterns for handling these delays using watermarks and grace periods in both PyFlink and Faust streaming applications.

Understanding Watermarks and Event Time

In stream processing, event time refers to when a record was actually generated at the source, while processing time refers to when the streaming engine receives it. When these diverge—due to network congestion, retries, or batch uploads—you need a mechanism to determine when windowed aggregations are complete.

Watermarks solve this by acting as a progress metric in event time. They are monotonically increasing timestamps that represent the point up to which the system assumes all data has been received. When a watermark t passes a window's end timestamp, that window is considered closed. Records arriving before the watermark (within the allowed lateness) are incorporated into the result; those arriving after are either dropped or routed to a side output.

Configuring Watermark Strategies in PyFlink

In 07-streaming/extras/pyflink/src/job/aggregation_job.py, the pipeline defines a watermark strategy that tolerates out-of-order events by subtracting a fixed duration from the current event timestamp.

The implementation uses WatermarkStrategy.for_bounded_out_of_orderness() to create a 5-second lag watermark (see line 58). This tells the engine that events may arrive up to 5 seconds late:

from pyflink.common.watermark_strategy import WatermarkStrategy
from pyflink.common.time import Duration
from pyflink.table import TableDescriptor, Schema, DataTypes

# Define a watermark strategy that allows 5 seconds of lateness

watermark_strategy = WatermarkStrategy \
    .for_bounded_out_of_orderness(Duration.of_seconds(5)) \
    .with_timestamp_assigner(lambda event, timestamp: event.event_timestamp)

# Apply to the source table with event_watermark derivation

t_env.create_temporary_table(
    "source_table",
    TableDescriptor.for_connector("kafka")
        .schema(
            Schema.new_builder()
                .column("event_timestamp", DataTypes.TIMESTAMP(3))
                .column_by_expression("event_watermark", "event_timestamp - INTERVAL '5' SECOND")
                .build()
        )
        .watermark("event_watermark", "event_timestamp - INTERVAL '5' SECOND")
        .build()
)

When the watermark advances past a window boundary, Flink finalizes the aggregation and emits the result. Events arriving within the 5-second tolerance are still processed; those arriving later are discarded.

Managing Late Arrivals with Faust Grace Periods

For Python-native streaming with Faust, 07-streaming/extras/python/streams-example/faust/windowing.py implements similar logic using the grace period parameter. This defines how long a window remains open after its end time to accept late-arriving records:

from datetime import timedelta
import faust

app = faust.App('taxi-analytics', broker='kafka://localhost:9092')

class Ride(faust.Record):
    event_timestamp: float
    passenger_count: int
    fare_amount: float

rides_topic = app.topic('rides', value_type=Ride)

# 1-minute tumbling window with 5-second grace for late data

windowed = rides_topic \
    .group_by(Ride.event_timestamp) \
    .tumble(timedelta(minutes=1), expires=timedelta(minutes=2), grace=timedelta(seconds=5)) \
    .reduce(lambda acc, x: acc + x.fare_amount, initial=0.0)

@app.agent()
async def process(windowed):
    async for key, total_fare in windowed.items():
        print(f'Window ending at {key}: total fare = {total_fare}')

The grace=timedelta(seconds=5) argument ensures that records arriving up to 5 seconds after the window closes are still included in the aggregation. Once the grace period expires, the window state is purged to prevent unbounded state growth.

Assigning Event Timestamps at the Source

Effective watermarking depends on accurate event-time extraction at the producer level. In 07-streaming/workshop/src/job/aggregation_job_demo.py and 07-streaming/extras/python/streams-example/faust/producer_taxi_json.py, the producer attaches a timestamp field to each message before sending it to Kafka:

import json
from datetime import datetime

def produce_event(producer, key, value):
    # Attach current event time in milliseconds

    value['event_timestamp'] = int(datetime.utcnow().timestamp() * 1000)
    producer.send(
        key=key, 
        value=json.dumps(value).encode('utf-8')
    )

# Example usage

produce_event(
    kafka_producer, 
    key='taxi-ride', 
    value={'passenger_count': 2, 'fare_amount': 15.5}
)

The event_timestamp field serves as the basis for both watermark generation and window assignment downstream.

Handling Data That Arrives Too Late

When records arrive after the watermark has advanced beyond the allowed lateness, they are considered too-late. In Flink, these can be routed to a side output for special handling or dead-letter queues rather than silently dropped:

# Conceptual Flink side output for late events

late_data_stream = result_stream.get_side_output(late_tag)
late_data_stream.add_sink(to_dead_letter_queue)

Faust similarly drops records that arrive after the grace period expires, though you can implement custom handlers in the agent logic to capture these events for audit trails or reprocessing pipelines.

Summary

• Extract reliable event-time fields from source records using producer-side timestamps like event_timestamp.
• Configure watermark strategies that reflect your maximum expected disorder, such as the 5-second lag implemented in aggregation_job.py.
• Define allowed lateness or grace periods that align with business SLAs to update closed windows with moderately late data.
• Route excessively late events to side outputs or dead-letter queues to ensure data quality without blocking the main pipeline.

Frequently Asked Questions

What is the difference between event time and processing time in streaming?

Event time is the timestamp when a record was created at the source (e.g., when a sensor reading occurred), while processing time is when the streaming engine actually receives and processes the record. Watermarks rely on event time to produce correct, reproducible windowed aggregations regardless of when data physically arrives at the pipeline.

How do I choose the right watermark lag interval?

Select a watermark lag that exceeds the maximum expected out-of-order delay in your source system. According to the DataTalksClub implementations, a 5-second bound works for moderately ordered streams, but high-latency sources (like mobile devices on unstable networks) may require 30 seconds or more. Monitor your late-data metrics and adjust accordingly.

What happens to records that arrive after the allowed lateness period?

Records arriving after the watermark has passed the window end time plus the allowed lateness are considered too-late. In PyFlink, these can be captured via side outputs; in Faust, they are typically dropped. You should route these records to a dead-letter queue or audit log for data quality monitoring.

Can I update results after a window has closed due to late data?

Yes, if you configure allowed lateness (Flink) or a grace period (Faust). For example, with a 5-second grace period, windows remain open for late updates during that interval. However, once the grace period expires and the watermark advances, the window state is purged and no further updates are possible—you would need to reprocess the affected window via a batch job instead.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}