Late data isn't an edge case — it's a property of every real production pipeline. This lesson teaches you the complete toolkit: watermarks, grace windows, idempotent reprocessing with Delta Lake, and correction strategies that cascade downstream without breaking anything.
It's 9:15 AM on a Monday. Your dashboards show a suspicious 40% drop in weekend sales. Your phone is buzzing. Your manager is asking if there was a system outage. You dig in and find that the sales data is fine — it just hasn't arrived yet. Mobile POS terminals from three retail locations had intermittent connectivity over the weekend, and they're still flushing their transaction queues. By noon, the numbers look normal. But you've already burned two hours investigating a non-problem, and someone may have made a bad business decision based on incomplete data.
This is the late-arriving data problem in its most common form, and it's not an edge case. It's Tuesday for most data engineering teams. IoT sensors buffer and batch. Mobile apps sync when connectivity returns. Third-party APIs have SLA windows measured in hours, not seconds. Upstream ETL jobs fail and retry. Every one of these scenarios produces data that arrives late relative to when the event actually occurred — and pipelines that don't account for this will silently produce wrong answers, often without any error or warning.
By the end of this lesson, you'll understand exactly how to architect pipelines that handle late data gracefully — not by ignoring the problem, but by building explicit strategies to detect, accommodate, and correct for it. We'll cover watermarking and grace windows in streaming systems, idempotent reprocessing patterns for batch pipelines, and correction strategies that let you update downstream aggregates without breaking everything that depends on them.
What you'll learn:
You should be comfortable with:
Before jumping to solutions, let's be precise about what we're actually dealing with. There are two related but distinct problems that often get conflated:
Late-arriving data is data that was generated at time T but doesn't appear in your pipeline until time T + Δ, where Δ is large enough to matter. The event happened on Saturday, but the record shows up in your Kafka topic on Monday morning.
Out-of-order data is data where the sequence of records in your pipeline doesn't match the sequence of when events actually occurred. Records with earlier event timestamps arrive after records with later timestamps. This can happen even without significant delay — a cluster of events from the same second might arrive in shuffled order.
Both problems stem from the fundamental distinction between event time (when something happened) and processing time (when your pipeline sees it). Any pipeline that aggregates by event time — daily sales totals, hourly API call counts, 5-minute sensor averages — is vulnerable.
Here's a minimal example to make this concrete. Imagine you're aggregating e-commerce orders by hour:
# What your pipeline sees, in arrival order:
orders = [
{"order_id": "A1001", "event_time": "2024-11-15 14:02:33", "amount": 89.99},
{"order_id": "A1002", "event_time": "2024-11-15 14:47:11", "amount": 124.50},
{"order_id": "A1003", "event_time": "2024-11-15 15:03:07", "amount": 34.00},
# Pipeline closes the 14:00 hour window here
{"order_id": "A1004", "event_time": "2024-11-15 14:38:44", "amount": 210.00}, # LATE
{"order_id": "A1005", "event_time": "2024-11-15 13:57:22", "amount": 67.25}, # VERY LATE
]
Without any late-data handling, A1004 and A1005 get either silently dropped or thrown into the wrong window. Your 14:00 hour shows $214.49 instead of $491.74. No error. No warning. Just a wrong number.
The core challenge is that you have to make a decision: when is a window "done"? If you wait forever, you have perfect accuracy but infinite latency. If you close immediately, you have low latency but miss late data. Every late-data strategy is just a different answer to this question.
Watermarks are the streaming world's answer to "when is a window done?" A watermark is a moving threshold that declares: "I believe all events with timestamps earlier than W have now arrived." When your watermark passes the end of a window, the system can safely emit that window's results.
Most streaming frameworks (Apache Flink, Spark Structured Streaming, Apache Beam) let you define a watermark as a lag behind the maximum observed event time. If the latest event you've seen has timestamp T, your watermark might be set to T minus 10 minutes. This means you're asserting that no event more than 10 minutes late will arrive.
In Spark Structured Streaming, it looks like this:
from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col, sum as spark_sum
spark = SparkSession.builder \
.appName("OrderAggregation") \
.getOrCreate()
# Read from Kafka
raw_orders = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "kafka-broker:9092") \
.option("subscribe", "ecommerce.orders") \
.load()
# Parse and cast your event schema
orders = raw_orders.selectExpr("CAST(value AS STRING) as json_value") \
.select(
col("json_value.order_id"),
col("json_value.amount").cast("double"),
col("json_value.event_time").cast("timestamp")
)
# Apply a 10-minute watermark
orders_with_watermark = orders.withWatermark("event_time", "10 minutes")
# Aggregate into hourly windows
hourly_totals = orders_with_watermark \
.groupBy(
window(col("event_time"), "1 hour"),
col("store_id")
) \
.agg(
spark_sum("amount").alias("total_revenue"),
count("*").alias("order_count")
)
# Write results
query = hourly_totals.writeStream \
.outputMode("append") \
.format("delta") \
.option("checkpointLocation", "/checkpoints/hourly_orders") \
.option("path", "/delta/hourly_orders") \
.start()
Notice outputMode("append"). This is critical. In append mode, Spark only emits a window's results once the watermark has passed that window's end time. You get one output per window, after the system believes all data has arrived. This is the conservative, accuracy-first approach.
This is where engineering judgment matters more than framework knowledge. Setting a watermark is essentially making an SLA commitment about your data sources. Ask these questions about each upstream source:
For our retail POS example, if terminals can buffer up to 48 hours of transactions offline, a 10-minute watermark will miss a huge chunk of legitimate late data. You'd need something like 72 hours to be safe — which means your hourly aggregates won't be emitted until 72 hours after the hour ends. That's probably not acceptable.
This is the fundamental tension, and it's why a single watermark is often the wrong architecture for systems with heterogeneous data sources.
A practical pattern is to emit multiple results per window: an early approximate result, possibly a mid-window update, and a final result after the watermark passes. This requires outputMode("update") or outputMode("complete") in Spark, or equivalent in Flink.
# Using update mode — emits a row whenever a window's aggregate changes
query = hourly_totals.writeStream \
.outputMode("update") \
.trigger(processingTime="1 minute") \
.format("delta") \
.option("checkpointLocation", "/checkpoints/hourly_orders_update") \
.option("path", "/delta/hourly_orders_update") \
.start()
In update mode, Spark emits an updated aggregate every trigger interval for any window that received new data. Your downstream system sees multiple rows for the same window, each more complete than the last. The trade-off: downstream consumers need to handle these updates — they can't just append everything to a report table.
Important: Update mode with Delta Lake is powerful here because Delta supports MERGE operations. Your downstream write can upsert on
(window_start, store_id)rather than append, ensuring the latest aggregate wins.
Some frameworks distinguish between the watermark (when to emit) and a grace window (how long to keep accepting updates after emission). Apache Flink's Table API makes this explicit:
-- In Flink SQL
SELECT
TUMBLE_START(event_time, INTERVAL '1' HOUR) as window_start,
store_id,
SUM(amount) as total_revenue,
COUNT(*) as order_count
FROM orders
GROUP BY
TUMBLE(event_time, INTERVAL '1' HOUR),
store_id
In Flink's DataStream API, you can configure allowed lateness directly on a window:
// Flink Java — shown here for conceptual clarity
DataStream<Order> orders = ...;
orders
.keyBy(order -> order.getStoreId())
.window(TumblingEventTimeWindows.of(Time.hours(1)))
.allowedLateness(Time.hours(72)) // Grace window
.sideOutputLateData(lateOutputTag) // Capture anything beyond grace window
.aggregate(new RevenueAggregator())
.addSink(new DeltaSink());
The allowedLateness period keeps window state alive after initial emission, and any late record that falls within this grace period triggers a correction event downstream. Records that arrive even after the grace period are diverted to a side output (a dead-letter channel) for separate handling.
This is a clean architecture: the main stream gets timely and corrected data, the side output captures anything too late to handle in-stream, and you have a separate batch job to decide what to do with the stragglers.
Not every pipeline is streaming. Many production data platforms use daily or hourly batch jobs, and they face the same late-arrival problem — just in a different shape. A batch job that runs at midnight to process "today's" data will miss events that arrive tomorrow morning.
The key discipline here is idempotency: your batch job should be able to re-run for any historical date and produce the correct result, replacing any previous output for that date without duplication or corruption.
The most reliable pattern for batch pipelines is processing data by event-date partitions and overwriting the entire partition on reprocessing. Let's build this out for a realistic scenario: a daily job that aggregates subscription events from a SaaS product.
import pendulum
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, count, countDistinct, lit
from pyspark.sql.types import StringType
def run_daily_aggregation(processing_date: str, lookback_days: int = 3):
"""
Aggregate subscription events for the given processing_date.
lookback_days: How many previous event_date partitions to reprocess.
This handles late-arriving data by re-computing recent windows.
"""
spark = SparkSession.builder \
.appName(f"SubscriptionEvents_{processing_date}") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.getOrCreate()
target_date = pendulum.parse(processing_date)
# We reprocess a rolling window of dates, not just today
dates_to_process = [
(target_date - pendulum.duration(days=i)).to_date_string()
for i in range(lookback_days + 1)
]
print(f"Reprocessing event_dates: {dates_to_process}")
# Read raw events — these are partitioned by event_date in the source
raw_events = spark.read \
.format("delta") \
.load("/delta/raw/subscription_events") \
.filter(col("event_date").isin(dates_to_process))
# Compute aggregates at the (event_date, plan_tier, event_type) grain
daily_agg = raw_events.groupBy(
col("event_date"),
col("plan_tier"),
col("event_type")
).agg(
count("*").alias("event_count"),
countDistinct("account_id").alias("unique_accounts"),
countDistinct("user_id").alias("unique_users")
).withColumn("computed_at", lit(processing_date).cast(StringType()))
# Write back — overwrite only the partitions we touched
daily_agg.write \
.format("delta") \
.mode("overwrite") \
.option("replaceWhere", f"event_date IN ({', '.join(repr(d) for d in dates_to_process)})") \
.save("/delta/aggregated/subscription_daily")
print(f"Successfully wrote {daily_agg.count()} rows for {len(dates_to_process)} dates.")
if __name__ == "__main__":
import sys
processing_date = sys.argv[1] if len(sys.argv) > 1 else pendulum.today().to_date_string()
run_daily_aggregation(processing_date, lookback_days=3)
The critical piece here is replaceWhere. Delta Lake's replaceWhere allows you to overwrite only specific partitions rather than the entire table. Without this, you'd either append duplicate data or overwrite the entire history. With it, you can safely re-run this job at any time and get a correct result.
The lookback_days=3 parameter is your operational grace window translated into batch terms. If your data SLA says events can be up to 3 days late, you reprocess the last 3 days every time you run. This costs more compute, but it's simple to reason about and operationally robust.
Warning: Be careful with
replaceWhereand non-partition columns. Delta will enforce that all data written satisfies the predicate, but it won't prevent you from accidentally filtering out data you meant to keep if your predicate doesn't match your partitioning scheme.
Reprocessing only works if your source data is complete for the windows you're reading. If you're reading from a system that prunes old data, or from a Kafka topic with a short retention window, you need a staging layer that persists raw events long enough to support reprocessing.
A common pattern is to write all raw events to a "landing zone" partitioned by arrival date (when your pipeline received the event), and separately track the event date as a column. Your aggregation job reads from the landing zone using a wide arrival-date window, but groups by event date. This way, an event that arrived today but has an event timestamp from three days ago will be included in the reprocessing of that three-day-old window.
# Landing zone schema: partitioned by arrival_date, contains event_date
# This ensures late arrivals are retained for reprocessing
raw_events_with_dates = raw_events.select(
col("event_id"),
col("account_id"),
col("user_id"),
col("event_type"),
col("plan_tier"),
to_date(col("event_timestamp")).alias("event_date"), # When it happened
to_date(col("ingestion_timestamp")).alias("arrival_date") # When we got it
)
# When reading for reprocessing, use a wide arrival window
# but filter/aggregate on event_date
recent_raw = spark.read \
.format("delta") \
.load("/delta/landing/subscription_events") \
.filter(
# Arrival window: capture anything that arrived recently enough to be relevant
col("arrival_date") >= "2024-11-12"
)
# Now your aggregation groups by event_date, not arrival_date
# Late arrivals get counted in the right event window
Reprocessing handles the upstream pipeline, but what about data that's already been published downstream? If your hourly sales table feeds a BI dashboard, a finance export, and three downstream data products, reprocessing the upstream aggregate is only half the battle.
The retraction pattern works by publishing a correction event alongside the corrected value. Instead of just overwriting, you publish a negative of the old value and a positive of the new value. Some systems (particularly event-driven ones) process these naturally. The Kafka Streams and Flink ecosystems have native support for changelog semantics where a null value for a key represents a retraction.
For SQL-based systems, you implement this with a correction_type column:
-- Your corrections table structure
CREATE TABLE subscription_daily_corrections (
correction_id STRING,
event_date DATE,
plan_tier STRING,
event_type STRING,
correction_type STRING, -- 'retraction' or 'assertion'
event_count BIGINT,
unique_accounts BIGINT,
unique_users BIGINT,
original_run_at TIMESTAMP,
correction_run_at TIMESTAMP
);
-- To compute final values, downstream queries sum retractions (negative) and assertions (positive)
SELECT
event_date,
plan_tier,
event_type,
SUM(CASE WHEN correction_type = 'assertion' THEN event_count
WHEN correction_type = 'retraction' THEN -event_count
END) AS final_event_count,
SUM(CASE WHEN correction_type = 'assertion' THEN unique_accounts
WHEN correction_type = 'retraction' THEN -unique_accounts
END) AS final_unique_accounts
FROM subscription_daily_corrections
GROUP BY event_date, plan_tier, event_type
This is powerful for audit trails. You can see every correction that was ever applied, when, and why. Finance teams love this because it gives them a full history of how numbers changed.
The downside: it's complex for consumers to handle correctly. Every downstream query needs to implement the retraction logic. If you have casual SQL users querying these tables, they'll get wrong answers unless you expose a view that does the retraction math for them.
For most production teams, the upsert approach is simpler and more durable. Rather than managing retraction events, you maintain a table of current-best-known values and MERGE corrections in. Downstream consumers always see the latest version.
from delta.tables import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.functions import current_timestamp
def apply_corrections(corrected_df, spark: SparkSession):
"""
Merge corrected aggregates into the published table.
Uses MERGE to upsert: insert new rows, update existing ones.
"""
target_table = DeltaTable.forPath(spark, "/delta/aggregated/subscription_daily")
# Add correction metadata
corrected_with_meta = corrected_df.withColumn(
"last_corrected_at", current_timestamp()
)
target_table.alias("target").merge(
corrected_with_meta.alias("source"),
"""
target.event_date = source.event_date
AND target.plan_tier = source.plan_tier
AND target.event_type = source.event_type
"""
).whenMatchedUpdate(set={
"event_count": "source.event_count",
"unique_accounts": "source.unique_accounts",
"unique_users": "source.unique_users",
"last_corrected_at": "source.last_corrected_at",
"computed_at": "source.computed_at"
}).whenNotMatchedInsertAll() \
.execute()
print(f"Correction merge complete. Rows affected: {corrected_df.count()}")
Delta Lake's time travel feature gives you an implicit audit trail here — every version of the table is preserved, so you can query what the table looked like before the correction:
# What did yesterday's numbers look like before this morning's correction?
pre_correction = spark.read \
.format("delta") \
.option("timestampAsOf", "2024-11-18 07:00:00") \
.load("/delta/aggregated/subscription_daily") \
.filter(col("event_date") == "2024-11-17")
post_correction = spark.read \
.format("delta") \
.load("/delta/aggregated/subscription_daily") \
.filter(col("event_date") == "2024-11-17")
pre_correction.join(post_correction, ["event_date", "plan_tier", "event_type"], "full") \
.select(
"event_date", "plan_tier", "event_type",
pre_correction["event_count"].alias("before"),
post_correction["event_count"].alias("after"),
(post_correction["event_count"] - pre_correction["event_count"]).alias("delta")
).show()
All of this is reactive. The proactive layer is instrumentation: understanding how late your data actually arrives, so you can tune your grace windows with evidence instead of guesses.
This job runs as a sidecar to your main pipeline, reading from the same raw events table and computing arrival delay statistics:
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
col, unix_timestamp, percentile_approx,
max as spark_max, avg, stddev, count, lit, current_timestamp
)
import pendulum
def compute_late_arrival_stats(spark: SparkSession, source_path: str,
lookback_days: int = 30) -> None:
"""
Compute late-arrival statistics for each data source.
Run this daily and store results to track SLA drift over time.
"""
cutoff = pendulum.now().subtract(days=lookback_days).to_date_string()
raw = spark.read \
.format("delta") \
.load(source_path) \
.filter(col("arrival_date") >= cutoff)
# Compute delay in seconds between event time and arrival time
with_delay = raw.withColumn(
"arrival_delay_seconds",
unix_timestamp(col("ingestion_timestamp")) - unix_timestamp(col("event_timestamp"))
)
# Aggregate by source system and day
stats = with_delay.groupBy("source_system", "arrival_date").agg(
count("*").alias("total_events"),
avg("arrival_delay_seconds").alias("avg_delay_seconds"),
stddev("arrival_delay_seconds").alias("stddev_delay_seconds"),
spark_max("arrival_delay_seconds").alias("max_delay_seconds"),
percentile_approx("arrival_delay_seconds", 0.50).alias("p50_delay_seconds"),
percentile_approx("arrival_delay_seconds", 0.95).alias("p95_delay_seconds"),
percentile_approx("arrival_delay_seconds", 0.99).alias("p99_delay_seconds"),
(count("*") - count(
col("arrival_delay_seconds").filter(col("arrival_delay_seconds") <= 300)
)).alias("events_beyond_5min")
).withColumn("computed_at", current_timestamp())
# Write to monitoring table
stats.write \
.format("delta") \
.mode("append") \
.save("/delta/monitoring/late_arrival_stats")
# Alert on SLA violations
violations = stats.filter(
col("p99_delay_seconds") > 3 * 24 * 3600 # p99 beyond 3 days
)
if violations.count() > 0:
violation_list = violations.select("source_system", "arrival_date",
"p99_delay_seconds").collect()
for row in violation_list:
print(f"ALERT: {row.source_system} on {row.arrival_date} "
f"has p99 delay of {row.p99_delay_seconds / 3600:.1f} hours")
# In production: send to PagerDuty, Slack, etc.
Run this nightly and you'll build a statistical baseline for every source system. When your grace window setting is "10 minutes" but the p99 delay for your mobile app events is 52 hours, you now have the numbers to justify changing it.
Tip: Track these stats over time with a time dimension. Late-arrival behavior is often seasonal — it gets worse during product launches, holidays, or after firmware updates push to IoT devices. Historical stats let you anticipate and prepare rather than react.
Let's put all of this together in a realistic project. You're building the data pipeline for a fitness app that tracks workout sessions. The app works offline and syncs when connectivity returns, so sessions recorded in the morning gym might not arrive until the user's phone connects to WiFi hours later.
Your pipeline needs to:
session_start_time (event time) and ingested_at (arrival time)Step 1: Set up your landing zone
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType, IntegerType, DoubleType
from pyspark.sql.functions import to_date, current_timestamp, col
spark = SparkSession.builder \
.appName("FitnessApp_Pipeline") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.getOrCreate()
# Define the schema for incoming workout sessions
session_schema = StructType([
StructField("session_id", StringType(), False),
StructField("user_id", StringType(), False),
StructField("device_id", StringType(), True),
StructField("workout_type", StringType(), True),
StructField("session_start_time", TimestampType(), False), # Event time
StructField("duration_seconds", IntegerType(), True),
StructField("calories_estimated", DoubleType(), True),
StructField("heart_rate_avg", IntegerType(), True),
])
def ingest_sessions(input_path: str, landing_zone: str):
"""Ingest raw sessions, adding ingestion metadata."""
raw = spark.read \
.schema(session_schema) \
.json(input_path)
enriched = raw \
.withColumn("ingested_at", current_timestamp()) \
.withColumn("event_date", to_date(col("session_start_time"))) \
.withColumn("arrival_date", to_date(current_timestamp()))
enriched.write \
.format("delta") \
.partitionBy("arrival_date") \
.mode("append") \
.save(landing_zone)
print(f"Ingested {enriched.count()} sessions")
Step 2: Daily aggregation with lookback
from pyspark.sql.functions import sum as spark_sum, count, countDistinct, avg
def build_user_daily_summary(processing_date: str, lookback_days: int = 7):
"""
Build user daily workout summaries, reprocessing the last N days
to capture late-arriving sessions.
"""
import pendulum
target = pendulum.parse(processing_date)
dates_to_reprocess = [
(target - pendulum.duration(days=i)).to_date_string()
for i in range(lookback_days + 1)
]
# Read from landing zone — wide arrival window to catch late data
# but aggregate by event_date
sessions = spark.read \
.format("delta") \
.load("/delta/landing/workout_sessions") \
.filter(col("event_date").isin(dates_to_reprocess))
daily_summaries = sessions.groupBy(
col("event_date"),
col("user_id"),
col("workout_type")
).agg(
count("session_id").alias("session_count"),
spark_sum("duration_seconds").alias("total_duration_seconds"),
spark_sum("calories_estimated").alias("total_calories"),
avg("heart_rate_avg").alias("avg_heart_rate"),
countDistinct("device_id").alias("unique_devices")
)
# Overwrite only the date partitions we're touching
daily_summaries.write \
.format("delta") \
.mode("overwrite") \
.option(
"replaceWhere",
f"event_date IN ({', '.join(repr(d) for d in dates_to_reprocess)})"
) \
.save("/delta/aggregated/user_daily_summaries")
print(f"Rebuilt summaries for {len(dates_to_reprocess)} dates.")
return daily_summaries
Step 3: Propagate corrections downstream
from delta.tables import DeltaTable
def propagate_to_weekly_rollups(corrected_dates: list[str]):
"""
After daily summaries are corrected, roll up to weekly metrics.
This is the downstream correction propagation step.
"""
# Find which ISO weeks are affected by our corrected dates
import pendulum
affected_weeks = set()
for d in corrected_dates:
dt = pendulum.parse(d)
affected_weeks.add(f"{dt.isocalendar()[0]}-W{dt.isocalendar()[1]:02d}")
print(f"Correcting weekly rollups for weeks: {affected_weeks}")
# Recompute weekly summaries from the (now-corrected) daily summaries
daily = spark.read \
.format("delta") \
.load("/delta/aggregated/user_daily_summaries")
from pyspark.sql.functions import date_trunc, weekofyear, year, concat_ws, lpad
weekly = daily \
.withColumn("iso_year", year(col("event_date"))) \
.withColumn("iso_week", weekofyear(col("event_date"))) \
.withColumn(
"iso_week_label",
concat_ws("-W", col("iso_year"), lpad(col("iso_week").cast(StringType()), 2, "0"))
) \
.filter(col("iso_week_label").isin(list(affected_weeks))) \
.groupBy("iso_week_label", "user_id", "workout_type") \
.agg(
spark_sum("session_count").alias("weekly_sessions"),
spark_sum("total_duration_seconds").alias("weekly_duration_seconds"),
spark_sum("total_calories").alias("weekly_calories")
)
# Merge corrections into the weekly table
weekly_table = DeltaTable.forPath(spark, "/delta/aggregated/user_weekly_summaries")
weekly_table.alias("target").merge(
weekly.alias("source"),
"target.iso_week_label = source.iso_week_label AND target.user_id = source.user_id AND target.workout_type = source.workout_type"
).whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()
Exercise checkpoints:
build_user_daily_summary to also emit a log of which sessions arrived more than 24 hours late?ingest_sessions twice with the same file? Is the pipeline idempotent? If not, how would you fix it?Mistake 1: Using processing time as a proxy for event time
This is the original sin. It's tempting to use ingested_at or created_at (the database timestamp from your source system) as your event time. Don't. These reflect when the record was written to the source, not when the underlying event occurred. A mobile app that buffers sessions offline will write everything to the backend with today's timestamp, even if the sessions happened last week.
Always capture the true event timestamp (button pressed, transaction initiated, sensor reading taken) and carry it through your pipeline explicitly.
Mistake 2: Setting grace windows based on the average case, not the tail
If your p99 delay is 48 hours but your grace window is 2 hours, you're missing 1% of events. For a pipeline processing 10 million events per day, that's 100,000 events per day falling outside your window. Always size grace windows against tail latencies, and monitor for shifts.
Mistake 3: Forgetting to propagate corrections downstream
You reprocess the source table, the daily aggregate looks correct, and you close the incident. Three days later, someone notices the weekly rollup still has the old number. Corrections cascade. Map your data lineage before an incident, not during one, so you know exactly what needs to be reprocessed when something changes.
Mistake 4: Append-only writes to aggregated tables
If your aggregation job appends rather than overwrites/merges, every reprocessing run adds a duplicate set of rows for the affected dates. Your downstream consumers start double-counting without any obvious error. Always use replaceWhere for partition-level overwrites or MERGE for row-level upserts in aggregated tables.
Mistake 5: Not tracking "as-of" snapshots for time-sensitive decisions
Some business decisions — invoicing, regulatory reporting, SLA measurements — are based on what the data showed at a specific point in time, not what we now know to be true. If you only keep the corrected version, you can't answer "what did the dashboard show the CFO at the end-of-month close?" Delta Lake time travel or an explicit snapshot table solves this, but you have to build it deliberately.
Debugging tip: When a window or daily partition looks wrong, the first thing to check is the delay distribution of its constituent events. Add a query that breaks down records by how many hours late they arrived — this will immediately tell you whether you missed data that fell outside your grace window or whether something else is wrong.
Late-arriving data is not an exception — it's a property of distributed systems under real-world conditions. The engineers who handle it well aren't the ones who have eliminated latency (no one does), but the ones who have made deliberate, explicit decisions about how much lateness to tolerate, how to detect violations of those tolerances, and how to correct for them when they occur.
Here's the framework to take away:
replaceWhere or MERGE semantics so re-running is safe. It should be safe enough to do automatically.Where to go next:
The goal is pipelines that your business can trust even on Monday morning, when the weekend data is still catching up.