Handling Late-Arriving Data in Pipelines

Every event has two times: when it happened, and when your pipeline found out. Data is late whenever the gap between them crosses a boundary you care about, such as the end of the day a report covers. Late data is not a rare fault; it is a permanent property of real systems, and the question is never whether to handle it but how late to wait and what to do afterwards.

Event time and arrival time

Spark's streaming guide draws the distinction clearly: event time is the time embedded in the data itself, as opposed to the time the engine receives it, and most analyses should be based on event time (Spark docs: Structured Streaming). Records arrive late for ordinary reasons: a device was offline, a batch upstream was delayed, a transaction finished long after it started. New York's taxi data documents one such mechanism: a store-and-forward flag marks trips held in vehicle memory because the vehicle had no connection to the server (TLC: data dictionary). The TLC itself publishes each month about two months later, to allow full vendor submissions (NYC TLC: trip record data).

How late is late? Measure it

Before choosing a strategy, measure. A trip is reported under its pickup day, but its record can only exist once the trip ends. This script asks how much of each day's data is missing if a daily report is built at midnight, or some hours later:

import urllib.request
from pathlib import Path
import pandas as pd

name = "yellow_tripdata_2024-01.parquet"
if not Path(name).exists():
    urllib.request.urlretrieve(f"https://d37ci6vzurychx.cloudfront.net/trip-data/{name}", name)
trips = pd.read_parquet(name, columns=["tpep_pickup_datetime", "tpep_dropoff_datetime", "store_and_fwd_flag", "fare_amount"])
trips.columns = ["pickup", "arrived", "stored", "fare"]          # a trip record can only arrive once the trip has ended
trips = trips[trips["pickup"].between("2024-01-01", "2024-01-31", inclusive="left") & (trips["arrived"] >= trips["pickup"])]
day_end = trips["pickup"].dt.normalize() + pd.Timedelta(days=1)   # midnight closing each trip's pickup day
late_by = trips["arrived"] - day_end                              # positive: arrived after its day closed

print(f"{len(trips)} trips on 30 days")
print(f"{'report built':<26}{'trips missed':>13}{'share':>8}{'fares missed':>14}")
for label, wait in [("at midnight", "0h"), ("1 hour after midnight", "1h"), ("3 hours after midnight", "3h"),
                    ("6 hours after midnight", "6h"), ("a day later", "24h")]:
    missed = late_by > pd.Timedelta(wait)
    print(f"{label:<26}{missed.sum():>13}{missed.mean():>8.2%}{trips.loc[missed, 'fare'].sum():>14,.2f}")

worst = (late_by > pd.Timedelta("0h")).groupby(trips["pickup"].dt.date).sum()
print(f"worst day at midnight: {worst.idxmax()} with {worst.max()} trips still in progress")
held = trips["stored"] == "Y"
print(f"trips held in vehicle memory before sending (store and forward): {held.sum()} ({held.mean():.2%})")

2862430 trips on 30 days
report built               trips missed   share  fares missed
at midnight                       23201   0.81%    713,092.53
1 hour after midnight              1396   0.05%     33,072.03
3 hours after midnight             1300   0.05%     27,851.60
6 hours after midnight             1258   0.04%     26,545.60
a day later                          10   0.00%      2,466.00
worst day at midnight: 2024-01-27 with 1834 trips still in progress
trips held in vehicle memory before sending (store and forward): 11021 (0.39%)

The shape of the result is typical of late data everywhere:

• Most lateness is short. A report built at midnight misses 23,201 trips, 0.81% of the month, worth over 700,000 dollars in fares. Waiting one hour cuts that to 1,396 trips, 0.05%.
• There is a long tail. Waiting three or six hours barely helps: about 1,300 trips still arrive later. Only a full day's wait brings the number down to 10. A small share of records is very late, and no reasonable delay catches them all.
• It varies by day. On the worst day, a Saturday, 1,834 trips were still in progress at midnight.
• Known mechanisms are only part of it. 11,021 trips, 0.39%, were held in vehicle memory before sending, a reminder that lateness has several independent causes.

Strategy 1: wait before closing the period

The cheapest fix is to run the daily job after a delay. The measurement tells you what the delay buys: one hour takes this data from 99.19% to 99.95% complete. Choose the delay from the lateness distribution and the freshness the business needs, and state the result: "yesterday's figures are available at 01:00 and are 99.95% complete".

Strategy 2: reprocess a window

Waiting cannot catch the tail, so pipelines restate recent periods. Each run recomputes the last few days, not just the last one, and replaces those partitions. With a two-day window, the trips that arrived up to a day late are picked up on the next run, leaving 10 of nearly three million. This only works if loads are idempotent, replacing a day's rows instead of appending to them; idempotency in data pipelines covers how. Tell report users that the most recent days are provisional and may change slightly.

Strategy 3: watermarks in streaming

Stream processors face the same problem continuously: how long to keep a time window open. A watermark is the answer: a threshold for how late data is expected to be. In Spark, late data within the threshold is guaranteed to be aggregated into its window, and data later than the threshold may be dropped, which lets the engine discard old state. The trade-off is explicit: a longer watermark means more complete results, more memory and later final answers. Records beyond the watermark should be routed to a side table for batch correction, not lost.

Strategy 4: late facts and missing dimensions

In a warehouse, lateness also affects joins. Kimball defines a late-arriving fact as one whose current dimensional context no longer matches: the fact must be given the dimension keys that were in effect when the event happened, not today's. With slowly changing dimensions, that means looking up the version valid on the event date. The opposite case also occurs: a fact arrives before its dimension row exists, such as a sale for a customer not yet loaded. Insert a placeholder dimension row with the business key, load the fact against it, and fill in the attributes when they arrive.

Detecting and monitoring lateness

• Store both times on every record: event time and load time. Without load time, lateness cannot be measured.
• Track the distribution, such as the 99th percentile of arrival delay per source, and alert when it shifts.
• Watermark incremental loads on arrival time, not event time; incremental data loads shows what goes wrong otherwise.
• Reconcile recent periods against the source after the reprocessing window closes.

How these choices fit into overall pipeline design is covered in ETL vs ELT.

Batch and real-time processing, stream processing concepts and data loading strategies are part of the Data Engineering course.

Sources

• Structured Streaming programming guide (Apache Spark documentation) — Apache Spark
• Late arriving facts (Kimball Group) — Kimball Group
• Data dictionary: yellow taxi trip records (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission