Streaming Late-Event Backfills With Event-Time Watermarks In Duckdb

The problem I ran into: “real-time” metrics that mysteriously move backward

I was building a small real-time analytics pipeline that computed a rolling metric like “orders per minute.” Everything looked great during tests—until I replayed production data with a tiny bit of network jitter.

What happened: some events arrived late (a.k.a. “late-arriving events”). Even though the stream kept going forward, the “real-time” chart occasionally jumped backward because late events were being included in already-emitted windows.

To make this precise, I implemented a very specific pattern:

Late-event backfill using event-time watermarks with DuckDB as an in-process SQL engine.

Here, event time means the timestamp embedded in each event (when it actually happened). Watermarks are a moving threshold that tells the system how far it believes event time has progressed, so it can safely finalize windows.

---

What I built: a tiny real-time window aggregator with watermarks

I used DuckDB to keep the example runnable and debuggable. The logic is:

1. Each event has event_time (the real time the event occurred).
2. We maintain a watermark based on the latest observed event time minus an allowed lateness.
3. We assign events to fixed windows (e.g., 1-minute windows).
4. When the watermark passes the end of a window, we consider it “closed.”
5. Late events that arrive after a window closes go into a backfill table, but only up to a configured backfill horizon.

The whole thing (working code)

import duckdb
from datetime import datetime, timedelta, timezone

con = duckdb.connect(database=":memory:")

# --- Schema ---
con.execute("""
CREATE TABLE events (
    order_id   VARCHAR,
    customer_id VARCHAR,
    event_time TIMESTAMP,
    arrival_time TIMESTAMP
)
""")

con.execute("""
CREATE TABLE window_metrics (
    window_start TIMESTAMP,
    window_end   TIMESTAMP,
    event_count  BIGINT,
    updated_at   TIMESTAMP
)
""")

con.execute("""
CREATE TABLE window_backfills (
    window_start TIMESTAMP,
    window_end   TIMESTAMP,
    late_event_count BIGINT,
    backfilled_at TIMESTAMP
)
""")

# --- Helpers ---
def window_bounds(ts: datetime, window_size_sec: int):
    epoch = int(ts.replace(tzinfo=timezone.utc).timestamp())
    w_start_epoch = (epoch // window_size_sec) * window_size_sec
    w_start = datetime.fromtimestamp(w_start_epoch, tz=timezone.utc)
    w_end = w_start + timedelta(seconds=window_size_sec)
    return w_start, w_end

def upsert_window_metric(window_start, window_end, delta_count, updated_at):
    # I do a simple upsert by updating existing rows, otherwise inserting.
    con.execute("""
    UPDATE window_metrics
    SET event_count = event_count + ?, updated_at = ?
    WHERE window_start = ? AND window_end = ?
    """, [delta_count, updated_at, window_start, window_end])

    affected = con.execute("""
    SELECT COUNT(*) AS cnt FROM window_metrics
    WHERE window_start = ? AND window_end = ?
    """, [window_start, window_end]).fetchone()[0]

    # If not present, insert it
    if affected == 0:
        con.execute("""
        INSERT INTO window_metrics(window_start, window_end, event_count, updated_at)
        VALUES(?, ?, ?, ?)
        """, [window_start, window_end, delta_count, updated_at])

def record_backfill(window_start, window_end, delta_count, backfilled_at):
    con.execute("""
    UPDATE window_backfills
    SET late_event_count = late_event_count + ?, backfilled_at = ?
    WHERE window_start = ? AND window_end = ?
    """, [delta_count, backfilled_at, window_start, window_end])

    affected = con.execute("""
    SELECT COUNT(*) AS cnt FROM window_backfills
    WHERE window_start = ? AND window_end = ?
    """, [window_start, window_end]).fetchone()[0]

    if affected == 0:
        con.execute("""
        INSERT INTO window_backfills(window_start, window_end, late_event_count, backfilled_at)
        VALUES(?, ?, ?, ?)
        """, [window_start, window_end, delta_count, backfilled_at])

# --- Configuration ---
WINDOW_SIZE_SEC = 60              # 1-minute windows
ALLOWED_LATENESS_SEC = 30        # watermark lags event time by 30s
BACKFILL_HORIZON_SEC = 180       # accept late backfills for up to 3 minutes past close

# --- Generate a realistic-ish stream ---
base = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)

# I simulate events arriving with different delays.
# event_time is when it happened; arrival_time is when it shows up to the system.
stream = [
    # On-time-ish events for 12:00 minute
    ("o1", "c1", base + timedelta(seconds=10), base + timedelta(seconds=12)),
    ("o2", "c2", base + timedelta(seconds=20), base + timedelta(seconds=22)),
    ("o3", "c3", base + timedelta(seconds=40), base + timedelta(seconds=45)),

    # Events for 12:01 minute arrive on time enough
    ("o4", "c4", base + timedelta(seconds=70), base + timedelta(seconds=74)),
    ("o5", "c5", base + timedelta(seconds=80), base + timedelta(seconds=86)),

    # Late events that belong to the 12:00 window (event_time=12:00:50)
    # but arrive during the processing of 12:01 and later.
    ("o6", "c6", base + timedelta(seconds=50), base + timedelta(seconds=115)),

    # Another late event but still within backfill horizon
    ("o7", "c7", base + timedelta(seconds=25), base + timedelta(seconds=160)),
]

# Sort by arrival_time to simulate streaming ingestion order
stream.sort(key=lambda r: r[3])

# --- Watermark state ---
max_event_time_seen = None
watermark = None  # dynamic

def current_watermark(max_event_time, allowed_lateness_sec):
    return max_event_time - timedelta(seconds=allowed_lateness_sec)

# --- Process the stream ---
for order_id, customer_id, event_time, arrival_time in stream:
    if max_event_time_seen is None or event_time > max_event_time_seen:
        max_event_time_seen = event_time
        watermark = current_watermark(max_event_time_seen, ALLOWED_LATENESS_SEC)

    # Determine the window for this event based on event_time
    w_start, w_end = window_bounds(event_time, WINDOW_SIZE_SEC)

    # Logic:
    # - If watermark has not yet passed the window end, treat as "on-time" for now.
    # - If watermark has passed window end, the window is closed; decide whether to backfill.
    window_closed = (watermark is not None) and (watermark >= w_end)

    if not window_closed:
        # Count normally
        upsert_window_metric(w_start, w_end, 1, updated_at=arrival_time)
    else:
        # Window is closed. Only backfill within horizon.
        # The idea: allow backfills only if the window end isn't too far behind the watermark.
        close_age = (watermark - w_end).total_seconds()
        if close_age <= BACKFILL_HORIZON_SEC:
            record_backfill(w_start, w_end, 1, backfilled_at=arrival_time)
            # Also apply the correction to the main metric so dashboards can reflect it.
            upsert_window_metric(w_start, w_end, 1, updated_at=arrival_time)
        else:
            # Drop too-late events (not part of the configured correction window)
            pass

# --- Inspect results ---
print("=== window_metrics (final + backfilled corrections) ===")
print(con.execute("""
SELECT window_start, window_end, event_count, updated_at
FROM window_metrics
ORDER BY window_start
""").fetchdf())

print("\n=== window_backfills (how many late events were corrected) ===")
print(con.execute("""
SELECT window_start, window_end, late_event_count, backfilled_at
FROM window_backfills
ORDER BY window_start
""").fetchdf())

---

Walkthrough: why this works (and what the tables mean)

1) I classify events by event time, not arrival time

Even though the stream is processed by arrival_time, the window assignment uses event_time.

That’s the core of stable event-time analytics: windows represent real-world time buckets, not ingestion time buckets.

2) I compute the watermark from the newest event time observed

When I see a new event_time, I update:

• max_event_time_seen
• watermark = max_event_time_seen - allowed_lateness

So if I allow 30 seconds lateness, the watermark always trails the “front” of the stream by 30 seconds.

3) I close a window when watermark passes its end

For a window [w_start, w_end):

• if watermark >= w_end, the system considers that window closed
• late events for closed windows are candidates for backfill

This is where “real-time” stops being misleading: windows won’t keep changing forever just because the network is messy.

4) I constrain backfills with a horizon

Without a horizon, late events could cause unbounded churn.

So I drop events when the window is too far behind the watermark:

• close_age = (watermark - w_end)
• accept if close_age <= BACKFILL_HORIZON_SEC

In practice, this gives dashboards a predictable “maximum correction delay.”

---

What happens when I run this?

In this simulated stream:

• o1, o2, o3 are for the 12:00 minute and arrive early enough → counted normally
• o6 and o7 belong to the 12:00 minute but arrive during later minutes → counted as backfills
• window_backfills shows how many corrections were applied per window

You’ll see the window_metrics.event_count for 12:00 increase after the late arrivals, but only because they were still within the backfill horizon. If I made lateness bigger than BACKFILL_HORIZON_SEC, those late events would be ignored and the 12:00 window would remain stable.

---

Data observability hook: detecting when “late correction” is getting worse

A really helpful operational metric is late correction volume per window. With the window_backfills table, you can monitor it like this:

df = con.execute("""
SELECT
  window_start,
  late_event_count,
  CASE
    WHEN late_event_count = 0 THEN 'ok'
    WHEN late_event_count <= 1 THEN 'minor'
    ELSE 'spike'
  END AS lateness_status
FROM window_backfills
ORDER BY window_start
""").fetchdf()

print(df)

If you graph late_event_count over time and it trends upward, you’ve likely got a real upstream delay issue (batching, queue backpressure, clock drift, etc.), and you can focus debugging where it matters.

---

Why I like this pattern for real-time analytics

I’ve built systems where “late event” handling was an afterthought, and charts slowly degraded into mystery. This approach makes three things explicit:

1. Event-time windows: stable meaning of “the minute.”
2. Watermark-based closing: bounded correctness, not infinite waiting.
3. Backfill horizon: controlled correction to avoid endless churn.

That combination is what turns “real-time analytics” from a live demo into something I can trust under real network and scheduling behavior.

---

Conclusion

I implemented late-event backfills using event-time watermarks with DuckDB, processing a stream in arrival order while assigning events to event-time windows. The key idea is that windows close when the watermark passes their end, and late events only correct already-closed windows within a configured backfill horizon—keeping dashboards stable while still recovering correctness.