Skip to main content

Late-Arriving Data and Watermarks

Event time says when something happened in the domain; arrival or processing time says when the pipeline observed it. They diverge during offline clients, network delays, retries, and outages. A watermark is a system's estimate of event-time progress, not proof that no older event can arrive.

Prerequisites​

Complete Incremental Processing. The fixture adds e-105, observed on October 3 but occurring on October 1.

Learning goals​

  • Distinguish event time, arrival time, windows, watermarks, and allowed lateness.
  • Choose whether late records update, retract, quarantine, or wait for a backfill.
  • Explain the completeness-versus-latency tradeoff.
  • Avoid presenting a watermark as a perfect knowledge boundary.

Worked example​

After the first two batches, the latest observed event time is October 2 at 10:00 UTC. This teaching policy places the watermark twelve hours behind that maximum. The later delivery has an event time before the watermark, so it follows the correction path.

import json
from collections import defaultdict
from datetime import datetime, timedelta
from decimal import Decimal
from pathlib import Path

def instant(value):
    return datetime.fromisoformat(value.replace("Z", "+00:00"))

root = Path("fixtures/data-engineering/pipelines")
base = []
for name in ("events-batch-1.jsonl", "events-batch-2.jsonl"):
    base.extend(json.loads(line) for line in (root / name).read_text().splitlines())
base = list({event["event_id"]: event for event in base}.values())
watermark = max(instant(event["occurred_at"]) for event in base) - timedelta(hours=12)
late = [json.loads(line) for line in (root / "events-late.jsonl").read_text().splitlines()]
late_ids = [event["event_id"] for event in late if instant(event["occurred_at"]) < watermark]

daily = defaultdict(Decimal)
for event in base + late:
    daily[event["occurred_at"][:10]] += Decimal(event["amount"])
totals = [(day, f"{amount:.2f}") for day, amount in sorted(daily.items())]
print(watermark.isoformat(), late_ids, totals)

The output identifies watermark 2026-10-01T22:00:00+00:00, late ID e-105, and corrected totals [('2026-10-01', '35.50'), ('2026-10-02', '10.00')]. The number twelve hours is a product policy for this fixture, not a universal default.

An update-producing system may revise the October 1 aggregate and label it provisional until its lateness horizon expires. An append-only consumer may require compensating records. Regulated reports may quarantine late changes for approval. Preserve the late record and make correction status observable in every case.

Limitations and common mistakes​

  • The maximum observed timestamp can jump because of a bad future timestamp; validate clocks and bounds.
  • Dropping late data silently makes low-latency output look more complete than it is.
  • Longer allowed lateness increases retained state and delays finality.
  • Processing-time windows are easier but answer a different question from event-time windows.
  • Different downstream uses may need different correction and finality policies.

Practice​

Write a policy for a daily report that may change for seven days. Specify provisional labels, correction frequency, consumer notification, and what happens after day seven.

Next steps​

Continue with Retries, Backfills, and Recovery to repair partitions safely after transient and historical failures.

Primary sources​

Checked 2026-10-15: Apache Beam windowing and late-data guide, Beam basics for watermarks and triggers, and Apache Flink event-time concepts.