ML data engineering
Orchestration, Idempotency, Backfills, and Recovery
Design run state, writes, checkpoints, backfills, catch-up, reconciliation, and auditable recovery.
By the end you can
- Model pipeline runs as explicit states with auditable inputs and outputs
- Design idempotent append, merge, and overwrite behavior
- Plan backfills as controlled migrations with bounded impact
- Recover from retries, partial writes, catch-up pressure, and dependency failure
Steps
Plan a backfill as a controlled migration
Backfills should not compete blindly with live production or overwrite history without review. Airflow conceded the point in its own architecture. Release 3.0.0, on 22 April 2025, shipped AIP-78, and the release notes record the change in one sentence: “Backfills are now fully managed by the scheduler, rather than being launched as separate command-line jobs.”
A replayed historical interval now follows the same scheduling, versioning and observability path as an ordinary run. It can be launched from the UI or the REST API. Replaying the past stopped being something that happens off to the side.
That move turns two of the steps below from advice into settings someone has to fill in. The operator chooses how many max active runs the backfill is allowed — the throttle that protects live freshness in step 3. The operator also picks among Missing Runs, Missing and Errored Runs, or All Runs, and that choice sets the blast radius estimated in step 2 and the scope of the version published in step 5. Astronomer's documentation for the feature spells out both dials.
A backfill launched as a command-line job outside the scheduler turns the same two dials. It just records neither of them anywhere a reviewer can find afterwards.
1. Freeze the specification
Record code, parameters, source versions, interval, and intended semantic change.
2. Estimate blast radius
Identify affected partitions, consumers, storage, compute, and downstream rebuilds.
3. Run in isolated output
Build a parallel version and throttle work to protect live freshness.
4. Compare and approve
Validate keys, aggregates, labels, features, slices, and consumer behavior against current history.
5. Publish with rollback
Switch references deliberately, notify consumers, and retain the previous version until confidence is established.
A backfill is a historical data release with a potentially enormous blast radius.
Failure is a normal execution path
Data pipelines fail. Sources are late, workers crash, schemas change, credentials expire, and downstream systems become unavailable. Reliability comes from predictable recovery rather than from pretending failures can be prevented.
An orchestrator schedules work and tracks dependencies. It does not make transformations correct automatically. Tasks need deterministic inputs, explicit outputs, safe retries, and observable state.
Idempotency means rerunning the same logical task with the same inputs creates no additional business effects. A task can reach that by overwriting one partition, by merging on stable keys, or by publishing a new immutable version and swapping a pointer atomically.
Backfills replay historical intervals under a named code and data version. They deserve separate capacity, validation and publication controls, because they can alter years of downstream state quickly.
Two of the most widely deployed pieces of pipeline infrastructure changed their own defaults on exactly these two points. Kafka 3.0 changed delivery guarantees on 21 September 2021. Airflow 3.0.0 changed catch-up on 22 April 2025. Both appear later in this lesson. Both moved a decision out of the application and into the platform, because the application kept getting it wrong.
Design every task as though it will run twice, stop halfway, and resume after its dependencies have changed.
Case
Kafka 3.0 turned idempotent producers on by default
Kafka moved the delivery guarantee out of the application and into its own defaults. Version 3.0, released 21 September 2021, implemented KIP-679. The project's own announcement states the result plainly: “Starting with 3.0, the Kafka Producer turns on by default idempotency and the acknowledgement of delivery by all of the replicas.”
Two defaults flipped in that release. enable.idempotence went from false to true. acks went from 1 to all.
The interesting part is what the stronger guarantee costs in configuration, because it is not free and it is not silent. Confluent's producer configuration reference lists the preconditions: max.in.flight.requests.per.connection at or below 5, retries greater than 0, and acks set to 'all'. A configuration that explicitly conflicts with them does not quietly degrade to the weaker guarantee. It raises a ConfigException.
That is the shape of a well-made default. The safe behaviour is on unless you say otherwise, and saying otherwise by accident fails loudly instead of shipping duplicates.
Visual
A dataset release moves through controlled states
Separating build from publication prevents consumers from reading partial outputs. Both major open table formats implement that separation the same way: the published version changes by pointer swap, not by mutating data in place.
The Apache Iceberg table specification says so in its section on optimistic concurrency: “An atomic swap of one table metadata file for another provides the basis for serializable isolation.” A writer whose base snapshot has gone stale does not overwrite the newer state. It retries against the new current version.
Delta Lake reaches the same guarantee from a different direction. It commits by atomically creating the next numbered JSON log record.
The 2020 Delta Lake paper also records what the word atomic cost on real storage. Amazon S3 offered no atomic put-if-absent and no atomic rename, so Databricks had to run a separate coordination service for log writes. Step 4 below is one line to write. Underneath it is an entire service someone had to build.
1. Plan
Resolve logical interval, source snapshots, parameters, code, and expected outputs.
2. Compute in isolation
Write temporary or versioned artifacts without changing the published view.
3. Validate
Run schema, quality, reconciliation, temporal, and consumer-specific gates.
4. Publish atomically
Expose the complete approved version through a transaction, snapshot, or pointer swap.
5. Record and reconcile
Store lineage, status, metrics, notifications, and recovery information.
Consumers should see either the previous good version or the complete new version, never an accidental mixture.
Comparison
Overwrite, merge, and append need different retry designs
Idempotency depends on output semantics rather than the scheduler’s retry button. Partition replacement, the simplest of the four patterns below, assumes atomic partition or snapshot publication, and the storage layer under most pipelines did not always supply it. The Delta Lake paper documented the earlier behaviour directly: a LIST issued after a PUT might not return the new object, so a job could replace a partition and then fail to see its own writing.
On 1 December 2020 AWS announced that Amazon S3 delivers strong read-after-write consistency automatically for all applications, including list operations, at no additional cost.
The consequence is visible in what got deleted afterwards. Apache Hadoop had shipped S3Guard, a DynamoDB-backed metadata store built purely to compensate for eventually consistent listings. Its documentation now reads: “Now that S3 is consistent, there is no need for S3Guard at all.” The code was removed from the source in 2022 under HADOOP-17409.
An entire subsystem existed because retry-safe overwrite was hard. It was deleted when the storage layer made it easy.
Partition replace
Recompute a bounded interval and replace it as one unit.
- Simple for deterministic daily or hourly builds
- Needs atomic partition or snapshot publication
- Late data may require reopening old intervals
- Example: rebuild one event-date partition
Keyed merge
Upsert records using stable business keys and change versions.
- Supports updates and CDC-style state
- Needs deterministic conflict resolution
- Deletes and tombstones require explicit handling
- Example: current account-state table
Append immutable version
Write a new dataset or run output without modifying prior artifacts.
- Strong audit and rollback properties
- Can increase storage and catalog volume
- Requires a separate publish pointer or catalog entry
- Example: versioned training snapshot
Unbounded append
Write events continuously to an append-only log.
- Natural for event streams and replay
- Downstream state still needs deduplication and checkpoints
- Corrections require compensating events or versions
- Example: raw click event topic
Key idea
Catching up can worsen an outage
After downtime, schedulers may launch many missed intervals simultaneously. The resulting load can overwhelm sources, storage, metadata services or downstream APIs, and extend the incident.
On Sunday 20 September 2015 that pattern ran to completion inside Amazon's own infrastructure. A brief network disruption sent DynamoDB storage servers in the US-East region back to the metadata service, and the retries then kept the overload alive after its original cause was gone: “Unavailable servers continued to retry requests for membership data, maintaining high load on the metadata service.” Customer error rates stabilised near 55% by 2:37am PDT.
Capacity could not simply be added into that load. AWS had to pause metadata-service requests at 5:06am PDT to shed the retry traffic first, and DynamoDB was restored at 7:10am PDT. The backlog was not a symptom to wait out. It was the thing that had to be stopped before recovery could begin.
The shape is general, not anecdotal. A 2022 study, Metastable Failures in the Wild, catalogues this incident as AWS3, 4.55 hours, sustaining effect "retry". It surveys 22 metastable failures at 11 organisations, and retries are the sustaining effect in more than 50% of them. The failure class itself had been described the year before, in a 2021 paper titled Metastable Failures in Distributed Systems.
So define concurrency limits, priority for current data, and a policy for obsolete intervals. Some late hourly outputs are unnecessary once a complete daily rebuild exists. Monitor queue age and when the backlog is expected to clear, not only active task count. A pipeline can look busy while freshness keeps deteriorating.
Recovery scheduling should protect current service while clearing historical debt at a controlled rate.
Case
Airflow 3 flipped catchup to False
Airflow changed its own default rather than leaving each deployment to discover the flood on its own. Version 3.0.0, released 22 April 2025, changed the [scheduler] catchup_by_default option from True to False. The release notes state it under Updated Scheduling Defaults: “Airflow 3.0 changes the default behavior for new DAGs by setting catchup_by_default = False in the configuration file. This means DAGs that do not explicitly set catchup=... will no longer backfill missed intervals by default.”
The direction of that change is the argument. A DAG that says nothing about catch-up used to replay every interval it had missed on activation. Now it replays none. The Airflow 2 behaviour is still there for anyone who wants it: set the option back to True, which Astronomer's upgrade guide documents as the restoration path.
A default is a bet about which mistake is more common. The project moved its bet.
Example
Retry bugs that create believable but wrong data
Each bug can leave the pipeline green after a successful retry. The second one is not cautionary folklore. It is documented semantics.
Kafka's design documentation says a consumer that saves its position before processing gets at-most-once, and one that saves it after processing gets at-least-once. The ordering of the checkpoint against the side effect is what chooses the failure mode. Google documents the same hazard from the other side: Pub/Sub may redeliver a message even after an acknowledgement request returned successfully, so subscribers must be idempotent.
Neither vendor treats duplicate or lost delivery as a defect to be fixed later. Both treat it as a property the application has to design around.
- Double append: a task writes rows, crashes before checkpointing, then appends the same interval again on retry.
- Early checkpoint: a consumer commits progress before the sink transaction and skips records after failure. Kafka's own design documentation describes exactly this ordering: “In this case there is a possibility that the consumer process crashes after saving its position but before saving the output of its message processing.”
- Partial overwrite: several partitions are replaced before a later partition fails, exposing mixed transformation versions.
- Nondeterministic tie: a deduplication task chooses an arbitrary first row, producing different survivors across retries.
- Stale dependency: a retried downstream task reads a newer reference table than the upstream output used originally.
Analogy
A pipeline run needs the guarantees of a bank transfer
Payments platforms made retried requests safe long before data pipelines borrowed the idea. The artefact they built is a client-supplied idempotency key. Stripe's API reference gives the mechanism: “Stripe’s idempotency works by saving the resulting status code and body of the first request made for any given idempotency key, regardless of whether it succeeds or fails.”
A later retry carrying the same key receives that stored response rather than creating a second charge. A retry whose parameters differ does not quietly do something new: Stripe compares the incoming parameters against the original and errors if they differ. Keys may be up to 255 characters, and may be pruned once they are at least 24 hours old.
The pattern is being standardised. An IETF Internet-Draft published on 15 October 2025 specifies the same idea as an HTTP header field, Idempotency-Key. A resource SHOULD answer a duplicate retry with the result of the previously completed operation, and a concurrent duplicate with a conflict error. Kafka's idempotent producer does the equivalent job inside the broker, deduplicating on a producer ID plus a sequence number.
All three implement the same two things: a logical operation identity that survives the retry, and a stored outcome the retry can be answered with. A data pipeline often transforms millions of records across many systems at once, so atomicity may come from publishing an immutable snapshot rather than from one distributed transaction. The identity and the recorded outcome are not optional in either setting.
Safe retry behavior requires a logical operation identity and a publication boundary consumers can trust.
Every run should have an auditable contract
Record the logical interval separately from the wall-clock time the job ran. A run launched today may rebuild data for last month.
Capture input versions, code, parameters, task attempts, output artifacts, validation results, publication status, and error classification. Retries should stay grouped under one logical run when they serve the same operation. This is the practical argument for AIP-78 pulling backfills into the Airflow 3.0.0 scheduler: a replayed interval that follows the same scheduling, versioning and observability path as an ordinary run leaves the same record behind. A command-line backfill leaves whatever its operator remembers.
Differentiate transient infrastructure failures from deterministic data failures. Retrying a broken schema indefinitely wastes resources and delays diagnosis. As the 2015 DynamoDB disruption showed at a larger scale, a retry that cannot succeed is not free while it waits.
Run metadata should answer three questions cleanly. What was intended? What actually happened? Which version became visible to consumers?
Operational metadata turns retries and backfills from guesswork into reconstructable system behavior.
Key takeaways
- Pipeline reliability assumes failures and designs predictable retry, recovery, and publication behavior. Kafka 3.0 and Airflow 3.0.0 both changed their own defaults rather than trusting each application to get this right.
- Idempotency is a property of logical effects and output semantics, not something an orchestrator supplies automatically: Stripe stores the first response under a key, Kafka deduplicates on producer ID plus sequence number, and at-most-once versus at-least-once turns purely on whether the checkpoint precedes the side effect.
- Compute, validation, and publication should be separate states, so consumers never see partial or mixed-version results. Iceberg publishes by an atomic swap of the table metadata file; Delta Lake by atomically creating the next numbered JSON log record.
- Partition replacement, keyed merge, immutable versioning, and unbounded append need different retry and correction designs, and the guarantees underneath them move: S3 gained strong read-after-write consistency on 1 December 2020, and Hadoop deleted S3Guard as a result.
- Backfills need frozen specifications, blast-radius analysis, isolated outputs, comparisons, capacity controls, and rollback. Airflow's AIP-78 makes two of those explicit: a max-active-runs throttle, and a choice among Missing Runs, Missing and Errored Runs, or All Runs.
- Run metadata should connect logical intervals, source versions, code, attempts, outputs, validation, and the version exposed to consumers. Catch-up must be scheduled deliberately: in the 20 September 2015 DynamoDB disruption, the retry backlog had to be paused at 5:06am PDT before capacity could be added at all.