What is the difference between event time and processing time?

Event time is when something actually happened (on the device or in the source system). Processing (or arrival) time is when the pipeline received or processed it. Late and out-of-order data happen when the two differ.

How do batch pipelines handle late-arriving data?

Partition by event time and recompute a lookback window of recent partitions each run (idempotently), sized from measured arrival delays, with periodic wider corrections and provisional labels on recent numbers.

What is a watermark with allowed lateness in stream processing?

The engine's estimate of how far event time has progressed. A window is finalised once the watermark passes its end plus the allowed lateness; events arriving later are dropped or sent to a side output.

Late and Out-of-Order Data — ETL and Data Pipelines

Scenario

Example company: a mobile app sends usage events. A job runs at 00:30 each night and counts yesterday's events by their event time (when the action happened on the phone). Phones that were offline upload their events when they reconnect, sometimes hours later.

What you see in production

Yesterday's active-user count is about 6% lower than the same number computed a week later.
Product managers screenshot the morning number, then see a different number on Friday, and stop trusting the dashboard.
Days near weekends and holidays are affected most, because people travel without a connection.

The naive code

# 00:30: count yesterday once and never look at it again
events = read_events(event_date=yesterday)
write_partition(yesterday, count_users(events))

Reproduce it

Each event has an event time (day of the action) and an arrival time (hours after midnight of that day when it reached the server).

One-shot daily counts vs a lookback windowTry it yourself

# (event_day, arrival_hours_after_that_day_starts)
events = [(1, 2), (1, 10), (1, 23), (1, 26), (1, 40), (1, 75),   # day 1: three arrive late
          (2, 5), (2, 20), (2, 30), (2, 49),
          (3, 1), (3, 12)]

def count_for(day, job_runs_at_hour):
    # the job sees events that arrived before it ran (measured from the start of `day`)
    return sum(1 for d, arrived in events if d == day and arrived <= job_runs_at_hour)

true_counts = {d: sum(1 for e, _ in events if e == d) for d in (1, 2, 3)}
print("true counts:     ", true_counts)

# Naive: count each day once, 30 minutes after it ends
naive = {d: count_for(d, 24.5) for d in (1, 2, 3)}
print("one-shot counts: ", naive)

# Lookback: every night, recompute the last 3 days (idempotent partition overwrite)
final = {}
for night in (2, 3, 4):              # job runs at the start of day `night` + 0.5h
    for day in range(max(1, night - 3), night):
        hours_since_day_start = (night - day) * 24 + 0.5
        final[day] = count_for(day, hours_since_day_start)
print("with 3-day lookback after night 4:", final)

# (event_day, arrival_hours_after_that_day_starts)
events = [(1, 2), (1, 10), (1, 23), (1, 26), (1, 40), (1, 75),   # day 1: three arrive late
          (2, 5), (2, 20), (2, 30), (2, 49),
          (3, 1), (3, 12)]

def count_for(day, job_runs_at_hour):
    # the job sees events that arrived before it ran (measured from the start of `day`)
    return sum(1 for d, arrived in events if d == day and arrived <= job_runs_at_hour)

true_counts = {d: sum(1 for e, _ in events if e == d) for d in (1, 2, 3)}
print("true counts:     ", true_counts)

# Naive: count each day once, 30 minutes after it ends
naive = {d: count_for(d, 24.5) for d in (1, 2, 3)}
print("one-shot counts: ", naive)

# Lookback: every night, recompute the last 3 days (idempotent partition overwrite)
final = {}
for night in (2, 3, 4):              # job runs at the start of day `night` + 0.5h
    for day in range(max(1, night - 3), night):
        hours_since_day_start = (night - day) * 24 + 0.5
        final[day] = count_for(day, hours_since_day_start)
print("with 3-day lookback after night 4:", final)

Output

true counts:      {1: 6, 2: 4, 3: 2}
one-shot counts:  {1: 3, 2: 2, 3: 2}
with 3-day lookback after night 4: {1: 5, 2: 3, 3: 2}

The lookback recovers most late events, but day 1 still shows 5, not 6: one event arrived 75 hours late, outside the 3-day window. Size the window from measured delays, and accept (or separately correct) the rare events beyond it.

Root cause

The job assumed all events for a day have arrived by 00:30 the next day. Event time (when it happened) and processing time (when it arrived) differ, and the gap has a long tail on mobile and IoT devices, partner file drops and retried webhooks.

The fix

Partition by event time and recompute a lookback window each run, for example the last 3 days, using idempotent partition overwrites. Choose the window from the measured arrival-delay distribution (for example the 99.9th percentile).
Mark recent days as provisional in reports, and "final" after the window closes.
For very late data beyond the window, either run a periodic wider correction (weekly over 30 days) or route it to a separate "late arrivals" table.
In streaming, the same idea is a watermark with allowed lateness: the engine waits a set time for late events before finalising a window.

Trade-offs

Lookback windows recompute data repeatedly, costing compute.
Longer windows make numbers more complete but keep them provisional longer.
Some metrics (billing) must never change after they are issued; those need an explicit cut-off and adjustment records instead of silent changes.

Detect it in production

Measure arrival delay (arrival time minus event time) and plot its percentiles.
Track how much each day's metric changes between its first and final computation.

Key points

Event time and arrival time differ; late data is normal.
Recompute a lookback window idempotently, sized from measured delays.
Label recent numbers provisional; publish corrections as adjustments when values must not change.