Measuring Freshness From Event Time, Not Arrival Time

A nightly load can finish on schedule, report success, and write the partition for the previous day, while the table it wrote is already stale. Every scheduled check passes, because the checks observe the pipeline. The source system’s replication may have fallen behind, so the extract read a consistent snapshot of a table that was itself hours old, and the result carries that age into the warehouse.

Nothing in the pipeline’s own telemetry shows this. The job ran, the row count is plausible, the partition exists. The only signal that distinguishes fresh data from data that merely arrived on time is a comparison between two clocks that live in different places: the clock of the event and the clock of the reader.

Two timestamps, and why the gap between them is not one number

Every record carries, explicitly or implicitly, a time at which the thing it describes happened. An order has a creation time; a sensor reading has a sample time; a page view has a client timestamp. The pipeline also knows a time at which it learned about the record: the ingestion timestamp, assigned by the writer or by the storage layer at commit.

The difference is the record’s lag. It is not a property of the pipeline alone, because it combines several delays that occur in series:

• Delay between the event and its capture by the source system, which is the source’s own write path.
• Delay between capture and extraction, which is the replication or CDC lag.
• Delay between extraction and the transform’s commit, which is the job’s runtime.
• Delay between commit and the reader’s query, which is the visibility delay of the table format.

Each of these can dominate in a different deployment, and a single freshness number computed as “now minus the newest event time in the table” is the sum of all four plus whatever the clock skew between source and reader contributes. That is why a lag figure should be reported as a distribution, not a maximum and not a mean.

Why the maximum is the wrong summary

Freshness measured as the age of the newest record answers “has anything arrived recently”, which is a question about the pipeline’s heartbeat, not about its completeness. A single event from a test account, or one record from the fastest-producing source in a union, keeps the maximum near zero while the bulk of the table is hours behind.

Freshness measured as the mean is worse in the other direction: a long tail of very late records drags the mean upward, so a table that is complete up to the last five minutes but contains a few records from three days ago reports as three days stale. The mean is dominated by the outliers it is meant to summarise.

The quantity that matters to a consumer is usually a high quantile of the age distribution: what fraction of records are older than the acceptable threshold, or equivalently, the age at which 95 or 99 percent of records are newer. A consumer that needs “most data from the last hour” is asking a question about a quantile. A consumer that needs “every record from the last hour” is asking a question about the maximum, and that consumer needs a completeness signal rather than a freshness one, because the maximum age is set by the single worst record in the table and is therefore a statement about the slowest source, not the fastest.

The two questions also have different remedies. A quantile failure indicates the pipeline is uniformly behind and the fix is throughput or scheduling. A maximum failure indicates one partition or one source is stuck and the fix is a targeted repair, which is why the diagnostic should always be per source and per partition rather than global.

Clock skew sets a floor on what can be measured

The lag calculation assumes the event timestamp and the reading clock are comparable. They are only comparable to within the skew between the machines that produced them, and that skew is not small in the general case.

Consumer hardware clocks drift, and a machine that has been running without synchronisation can be seconds or minutes off. Network time synchronisation bounds the error but does not eliminate it, and the bound depends on the synchronisation interval and on the network path. A source system whose clock is two minutes ahead produces events that appear to have negative lag when read by a pipeline whose clock is correct, and a naive check that alerts on negative lag will fire on every batch from that source.

Two consequences follow. First, freshness thresholds below the plausible skew are not measurable, so a target of “under ten seconds” requires knowing that both clocks are synchronised to better than ten seconds, which is a statement about the deployment and not about the pipeline. Second, negative lag should be treated as a data quality signal in its own right rather than clamped to zero. A record whose event time is in the future is either a clock problem or a producer that writes a scheduled time rather than an observed one, and both are worth knowing about. Clamping hides them.

Event time can also be derived rather than observed. A record whose timestamp is assigned by the producer at the moment it batches a write, rather than at the moment the event occurred, carries the batching interval as a systematic offset. The offset is not random, so it does not average out, and it inflates every lag measurement by the same amount.

Partitioning turns staleness into a discrete event

Most warehouses partition by event date or hour, and that changes the shape of the problem. Freshness is then not a continuous quantity but a set of partition states: the partition for hour $h$ is either present, absent, or present with fewer rows than expected.

The advantage is that staleness becomes a predicate that can be checked without scanning data, since the set of partitions is metadata. A check that asks whether yesterday’s partition exists is a metadata read costing milliseconds, and it catches the common failure of a scheduler that did not fire.

The disadvantage is granularity. A partition can exist and be empty, or exist and contain a tenth of its usual rows, and the metadata check passes in both cases. This is why partition presence and partition volume are separate checks: presence is a scheduling signal, volume is a data signal, and they fail for different reasons. The volume check is also where partition-level row counts as a validation signal begins, since the same metadata is used for both.

Late data interacts badly with date partitioning. A record whose event date is three days ago, arriving today, must be written into the partition for three days ago, which means rewriting a partition that was already complete and possibly already read by consumers. Table formats handle this with a commit that replaces the partition’s file list, but the consumer that read the old version has already seen a table that was missing that record, and its result is now unreproducible from the current table state. Late writes therefore make freshness and reproducibility pull in opposite directions: a pipeline that refuses late writes is reproducible and incomplete, and one that accepts them is complete and non-deterministic for earlier readers.

What a watermark actually promises

Streaming systems express freshness through watermarks, and the promise a watermark makes is narrower than it is often read to be.

A watermark is an assertion that no future record will have an event time earlier than the watermark’s value, or more precisely in the common implementation, that records earlier than it are unlikely and will be treated as late. The assertion is a heuristic derived from observed event times and configured lateness bounds. It is not a guarantee about the source, and a source that goes silent does not produce a watermark that advances: a watermark generator that waits for evidence of progress will stall when the evidence stops, which is exactly the situation a freshness check is meant to detect.

# Age of each record, and the quantiles that describe the table's staleness.
def lag_summary(rows, now):
    lags = sorted(now - r.event_time for r in rows)   # timedeltas, ascending
    if not lags:
        return None
    n = len(lags)
    def q(p):                                         # nearest-rank quantile
        return lags[min(n - 1, int(p * n))]
    return {
        "count": n,
        "p50": q(0.50),
        "p95": q(0.95),
        "p99": q(0.99),
        "max": lags[-1],
        "negative": sum(1 for r in rows if r.event_time > now),
    }

The consequence is that a stalled watermark and a slow source look identical from the watermark’s position alone: both report a constant lag. Distinguishing them requires a second signal, the rate at which records arrive, which is a processing-time measurement and therefore immune to the event-time stall. The two signals together separate “no new events exist” from “new events exist but are not arriving”, and only the second is a fault. The same period-over-period reasoning used in seasonal baselines applies: what matters is the change against an expectation, not the level.

For a freshness check, the practical rule is that a watermark is a lower bound on completeness and never an upper bound on staleness, so it can be used to say “at least this much has arrived” and not to say “nothing older will arrive”.

Reporting freshness so that it can be acted on

A freshness metric is only useful if it names the population it describes. The three populations that differ most are the whole table, the newest partition, and the set of records written since the last check. Their lags can differ by orders of magnitude, and a single number that mixes them will be dominated by whichever is largest.

The arrangement that survives contact with real incidents reports, per source and per partition: the newest event time, the count of records in the last hour of event time, and the fraction of records older than the threshold. The first catches a dead source, the second catches a partial load, and the third catches a table that is receiving data steadily but has stopped making progress on the older partitions. None of the three is derivable from the others, which is the reason all three are needed rather than one well-chosen aggregate.

The final constraint is that the check must not depend on the pipeline it monitors. A freshness check that runs inside the same job as the load fails at exactly the moment the load fails, and reports nothing. Running it from a separate process that reads only metadata, on a schedule set independently, is what makes the check’s silence meaningful.