Event Time vs Processing Time: Know Which Clock Your Numbers Are On

On Tuesday morning the dashboard says Monday had 48,210 orders. On Wednesday morning it says Monday had 49,877. Nobody backfilled anything and no job failed. Each day someone asks whether the numbers are “still settling”, and nobody can say what that means or when it stops.

When a past day’s total keeps moving and nobody can explain why, the usual cause is that the pipeline is using one clock while the report is labelled with another.

Every record has at least three times

Clock	Meaning	Who sets it	Can you trust it?
Event time	When the thing happened in the real world	The producer: a device, app, or source database	Mostly. Device clocks drift, and some events arrive dated 1970 or next year
Ingestion time	When it reached your platform (broker, landing zone, bronze)	Your platform	Yes. It’s your clock, and it only goes forward
Processing time	When a particular job run read it	Whatever machine runs the job	It’s accurate, but it’s a property of the run, not the data

Batch pipelines add a fourth clock: the logical date of the run, meaning the slice of time the orchestrator asked the job to handle. A rerun of “the 1st of October” on the 9th has a logical date of the 1st and a processing time of the 9th.

These would all match in a perfect system. In practice they drift apart for ordinary reasons: mobile apps buffer events while offline, SaaS sources export in hourly batches, upstream systems have outages and then flush their backlog, and retries delay delivery. The gap is normally seconds. Sometimes it’s days.

The bugs, one clock at a time

Reporting processing time as if it were event time

This query looks like “orders per day”:

SELECT to_date(_ingested_at) AS order_date, count(*) AS orders
FROM bronze.orders
GROUP BY 1

It actually counts orders received per day, and the label is wrong. On a normal day the two are close enough that nobody notices. After an upstream outage that flushes 30,000 orders the next morning, Monday looks like a disaster and Tuesday looks like a record, and both days get an incident review.

Selecting new data by event time

The opposite mistake happens in incremental loads:

SELECT * FROM bronze.orders
WHERE event_ts > '${last_run_max_event_ts}'

Any event that arrives after a later event has already been processed will never be picked up, because its event time is behind the cursor. The offline phone in the figure is one example. Nothing errors and the data simply never arrives. That’s the main subject of late-arriving data.

Deriving “yesterday” from the processing clock

WHERE event_date = current_date() - 1 lets when the job ran decide which data it processes. A retry that lands after midnight processes the wrong day, and a backfill run on the 9th processes the 8th, however many times you click rerun. The logical date has to come from the orchestrator as a parameter. This is also the most common way jobs lose idempotence.

Time zones

Event time is only meaningful together with a zone, and “day” is a business definition, not a property of the timestamp. Three things go wrong regularly:

• to_date(ts) in Spark uses spark.sql.session.timeZone. Two clusters with different session time zones produce different daily totals from the same table and the same SQL.
• Sources send local time without an offset. Twice a year, daylight saving creates an hour that happens twice and an hour that never happens.
• The business day is in local time (Sydney trading, or a US retail calendar) while storage is in UTC. Midnight UTC is 10 or 11 in the morning in Sydney, so a “daily” UTC partition splits a Sydney trading day in two.

Normalise to UTC at ingestion, keep the original offset, and derive business dates explicitly:

from pyspark.sql import functions as F

spark.conf.set("spark.sql.session.timeZone", "UTC")   # pin it; never inherit it

events = (
    raw
    .withColumn("event_ts", F.to_utc_timestamp("event_local_ts", F.col("source_tz")))
    .withColumn("event_date_utc", F.to_date("event_ts"))
    .withColumn("business_date",
                F.to_date(F.from_utc_timestamp("event_ts", "Australia/Sydney")))
)

Device clocks

Event time comes from a clock you don’t own. Phones with the wrong date, embedded devices that reset to the epoch, and servers whose NTP has drifted all produce events “from the future” or from 1970. Check event time against ingestion time, which you do control:

skew = F.col("event_ts").cast("long") - F.col("_ingested_at").cast("long")

events = events.withColumn(
    "event_time_quality",
    F.when(skew > 300, "future")                                   # > 5 min ahead of arrival
     .when(F.col("event_ts") < F.lit("2015-01-01"), "implausible")
     .otherwise("ok"))

Decide per source what happens to future and implausible rows: substitute ingestion time, quarantine them, or keep them and exclude them from reporting. Write the decision down. Silently converting them is how a device fleet with a bad firmware update ends up all counting on the same day in 2036.

The rule: select by arrival, report by occurrence

Most of this comes down to one rule that you can enforce in code review:

In a batch pipeline that looks like this. Find what arrived since the last run, work out which event dates those arrivals affect, and recompute exactly those dates:

from pyspark.sql import functions as F

def run(spark, prev_hw: str, hw: str):
    arrived = (spark.read.table("bronze.orders")
               .where((F.col("_ingested_at") > prev_hw) &
                      (F.col("_ingested_at") <= hw)))

    touched = [r.d for r in
               arrived.select(F.to_date("event_ts").alias("d")).distinct().collect()]
    if not touched:
        return

    in_list = ", ".join(f"DATE'{d}'" for d in touched)

    daily = (spark.read.table("silver.orders")
             .where(f"order_date IN ({in_list})")
             .groupBy("order_date")
             .agg(F.count("*").alias("orders"), F.sum("amount").alias("revenue")))

    (daily.write.format("delta").mode("overwrite")
        .option("replaceWhere", f"order_date IN ({in_list})")
        .saveAsTable("gold.daily_orders"))

A record that arrives ten hours (or ten days) late adds its event date to touched, and that day is rebuilt from scratch. Rebuilding a day is idempotent, and the cursor is on ingestion time, so nothing is skipped. In practice you’d also cap touched (if a backfill touches 900 dates, you want an alert rather than a silent 900-day rebuild) and record which days were restated so that consumers know.

The same split applies to storage layout:

• Bronze: partition or cluster by ingestion date. It’s append-only, you can replay “everything we received on the 3rd”, and late events never land in an old partition.
• Silver and gold: partition or cluster by event or business date, which is how people filter their queries.

In streaming, the engine needs to be told

Streaming engines expose the same choice directly. In Spark Structured Streaming, a window on a timestamp column from the data is an event-time window:

from pyspark.sql import functions as F

per_hour = (
    orders_stream
    .withWatermark("event_ts", "2 hours")
    .groupBy(F.window("event_ts", "1 hour"), "store_id")
    .agg(F.count("*").alias("orders"))
)

The window belongs to the event. The watermark is the part that admits processing time exists: it tells Spark how long to wait for stragglers before it finalises a window and discards the state. Choosing that value and dealing with events beyond it is the subject of the next post in the series.

A column convention that prevents arguments

Name every clock explicitly and in the same way across tables, so a reader never has to guess what timestamp or date means:

Column	Meaning
event_ts	When it happened, in UTC
event_tz / event_offset	The source’s original zone or offset
business_date	The reporting day, derived with the business’s zone and calendar
_ingested_at	When the platform received it
_processed_at	When this row was last written by a job
_run_date	The logical date of the run that wrote it

The underscore prefix marks platform metadata as opposed to business data. Consumers learn quickly that _ columns are for lineage and debugging and shouldn’t appear in a report axis.

On Monday

• Pin spark.sql.session.timeZone explicitly in every job and notebook.
• Search for to_date(_ingested_at and current_date() in transformation code. Each hit is a clock mix-up waiting for an outage to expose it.
• Move incremental cursors from event time to ingestion time.
• Add an event-time quality flag comparing event time with ingestion time.
• Write a one-line definition of business_date and the zone it uses, and put it in the table comment.

Monday’s number will keep changing for a while, because events about Monday keep arriving after Monday. The goal is to make that predictable: you know why it changes, by how much, and when it stops. That’s where late-arriving data comes in.