Late-Arriving Data: Deciding When a Day Is Actually Done

The previous post ended with Monday’s order count still changing on Wednesday. The fix there was to group by event time and select by ingestion time, so that late events land on the right day. That leaves three questions it didn’t answer: how late can events be, how long should you wait for them, and what happens to the ones that arrive after you stop waiting?

Most teams never answer those questions on purpose. The answers end up being whatever the pipeline happens to do: a lookback of one day because the first version had one, a streaming watermark of ten minutes copied from a tutorial, and a finance report that silently disagrees with the dashboard every month-end.

Lateness is a distribution: measure it

“Late” means the gap between event time and ingestion time. Every source has its own distribution of that gap, with a long tail. You can’t pick a sensible waiting policy without seeing that tail.

The most useful view is a completeness curve: for event days old enough that their stragglers have certainly landed, what fraction of each day’s final rows had arrived by D+0, D+1, D+2, and so on?

WITH settled AS (               -- only days old enough that the tail has landed
  SELECT datediff(to_date(_ingested_at), to_date(event_ts)) AS days_late
  FROM bronze.orders
  WHERE event_ts >= date_sub(current_date(), 120)
    AND event_ts <  date_sub(current_date(), 30)
)
SELECT days_late,
       count(*) AS rows,
       round(100 * sum(count(*)) OVER (ORDER BY days_late)
                 / sum(count(*)) OVER (), 3) AS pct_complete
FROM settled
GROUP BY days_late
ORDER BY days_late

The window that excludes the last 30 days matters. If you include recent days, their late rows haven’t arrived yet, and the curve looks better than reality. An example of what comes back (your numbers will differ, which is the point of running it):

Days late	Cumulative % of a day’s rows
0 (same day)	91.8%
1	98.9%
2	99.6%
3	99.93%
7	99.99%
30+	100%

That table turns an argument into a decision. With a three-day lookback, a published day is 99.93% complete, and the 0.07% that misses it is the volume you need a separate plan for. Run the same query per source system. One mobile client with a bad offline queue can account for the whole tail.

Hours and minutes work the same way for streaming sources: unix_timestamp(_ingested_at) - unix_timestamp(event_ts) with percentile_approx at 0.5, 0.99, and 0.999.

Three decisions, written down

If the third answer is “we don’t know”, the actual answer is either dropped silently or restated silently, depending on how the code happens to be written. Neither is acceptable for a revenue number.

Batch: rolling recompute

The simplest batch policy that works: on every run, rebuild the last N event days from scratch, where N comes from the completeness curve.

from datetime import date, timedelta

LOOKBACK_DAYS = 3        # from the completeness curve: 99.93% at D+3

def run(spark, logical_date: str):
    end = date.fromisoformat(logical_date)
    start = end - timedelta(days=LOOKBACK_DAYS)
    window = f"order_date BETWEEN DATE'{start}' AND DATE'{end}'"

    daily = spark.sql(f"""
        SELECT order_date, store_id,
               count(*)    AS orders,
               sum(amount) AS revenue,
               max(_ingested_at) AS data_as_of
        FROM silver.orders
        WHERE {window}
        GROUP BY order_date, store_id
    """)

    (daily.write.format("delta").mode("overwrite")
        .option("replaceWhere", window)
        .saveAsTable("gold.daily_orders"))

This is idempotent because replaceWhere replaces a fixed slice. It’s also easy to reason about: any day within N of today is provisional, and anything older is final. The cost is recomputing N days on every run, which is cheap for daily aggregates and expensive for large fact tables. For those, use the “recompute only the touched days” variant from the previous post: find the event dates in newly ingested rows and rebuild only those, capped at the lookback so a stray 2019 event doesn’t reopen 2019.

Tell consumers what’s final

A number that might still change needs a label saying so. Two columns are enough:

• data_as_of: the latest ingestion time included in the row (as computed above).
• is_final: order_date < logical_date - LOOKBACK_DAYS.

Dashboards can then show provisional days differently, such as a hatched bar or a footnote, and the “the numbers changed!” message becomes a known property of the report rather than an incident.

After the books close

Some consumers, finance in particular, can’t accept a closed period changing. For them, events that arrive after the finality point shouldn’t modify the closed day at all. They go into an adjustments table dated by arrival, attributed to the original event date, and reported in the current open period. That’s how accounting has handled late invoices for a few hundred years. The data platform doesn’t need a better answer, only the same one implemented on purpose.

Streaming: watermarks are the same policy, enforced by the engine

In Spark Structured Streaming, a watermark is the waiting policy expressed in code. Spark tracks the maximum event time it has seen, subtracts the delay you give it, and treats that as the point before which it no longer expects data.

Four details matter in production:

• The guarantee only goes one way. Spark guarantees it won’t drop data that’s within the delay. It does not guarantee it will drop data beyond it, because rows are evaluated against the watermark as of the current batch. Don’t build logic that depends on late rows definitely being excluded.
• Output mode changes latency. In append mode a window is emitted only once the watermark passes its end, so a one-hour window with a two-hour delay appears up to three hours after its first event. update mode emits partial results as they change, which suits sinks that can upsert.
• Several inputs share one watermark. When a query has more than one watermarked input (for example a stream-stream join), Spark uses the minimum by default (spark.sql.streaming.multipleWatermarkPolicy = min), so the slowest stream holds back state cleanup for all of them.
• Dropped rows are counted, not kept. Each stateful operator reports numRowsDroppedByWatermark in the query progress. There’s no side output containing the rows. Monitor the count:

progress = query.lastProgress or {}
dropped = sum(op.get("numRowsDroppedByWatermark", 0)
              for op in progress.get("stateOperators", []))
if dropped:
    emit_metric("orders_per_hour.rows_dropped_by_watermark", dropped)

The design that keeps streaming honest is to never apply the watermark to your only copy of the data. Land every raw event in bronze unconditionally, with no watermark and no state. Run the watermarked aggregation as a fast, slightly approximate view for operational dashboards, and have the batch rolling recompute from bronze produce the numbers that are reported. The watermark controls how much state the stream keeps in memory. It shouldn’t decide which events your business ever sees.

Late-arriving dimensions (facts that arrive first)

The same problem appears in dimensional models in reverse. An order arrives for customer_id = 88412, but the customer record from the CRM hasn’t been loaded yet because it’s on a slower feed. The fact is on time and the dimension is late.

Option	What happens	Problem
Inner join, drop the fact	The order disappears from revenue	Silent data loss
Map to an “Unknown” member (-1)	The order counts, but under Unknown	The link is lost forever unless you reprocess the fact
Inferred member	Create a placeholder dimension row for 88412 now, and fill it in when the real record arrives	A little more logic in the dimension load

Inferred members keep the fact’s surrogate key stable, so no fact reprocessing is needed. Before loading facts, insert any business keys the dimension hasn’t seen:

MERGE INTO gold.dim_customer AS d
USING (SELECT DISTINCT customer_id FROM silver.orders_batch) AS f
ON d.customer_id = f.customer_id
WHEN NOT MATCHED THEN INSERT
  (customer_sk, customer_id, name, segment, is_inferred, valid_from, valid_to, is_current)
VALUES
  (xxhash64(f.customer_id), f.customer_id, 'Unknown (pending)', 'Unknown',
   true, TIMESTAMP'1900-01-01', TIMESTAMP'9999-12-31', true)

When the real customer record arrives, the dimension load updates the inferred row in place (WHEN MATCHED AND d.is_inferred THEN UPDATE ... is_inferred = false) instead of creating a new version. The placeholder was never a true historical state, so it shouldn’t appear in history. valid_from = 1900-01-01 means facts from before the record existed still find it. Track a count of is_inferred = true rows older than a day. A growing count means the dimension feed is broken, not just slow.

And when a fact arrives late into a Type 2 dimension, look up the version that was valid at the fact’s event time, not the current one. Otherwise a February order gets attributed to the segment the customer moved to in October.

On Monday

• Run the completeness curve for your three most important sources and read the tail.
• Set each table’s lookback from the curve, and add data_as_of and is_final columns.
• Write the after-policy rule for finance-facing tables: restate or adjust.
• Make sure no watermark sits between a source and your only stored copy of its data.
• Alert on numRowsDroppedByWatermark and on old inferred dimension members.

Data will always arrive late. What you control is whether you’ve decided what happens to it, or whether the code decides by accident.