A pipeline can finish successfully and still produce the wrong result after a retry. Imagine an orders load that writes its data, then fails during a later task. Repeating the load with an append can add the same orders again. Replacing existing rows without checking their version can introduce a different problem: an older delivery can overwrite a newer order status.
In this article, I walk through a small PySpark and Delta Lake example that handles repeated deliveries and older updates. The target stores the latest state of each order. Reprocessing the same input should leave that state unchanged. This property is called idempotency.
Start with the meaning of a duplicate
Consider these incoming records:
|order_id|source_version|status |
|--------|--------------|-------|
|101 |1 |Created|
|101 |2 |Shipped|
|101 |2 |Shipped|
|102 |1 |Created|
The two identical records for order 101 at version 2 are duplicates. Versions 1 and 2 represent different states of the same order. Because the target stores current state, the intended result is one row for order 101 with status Shipped and one row for order 102 with status Created.
For this example, the source must provide a non-null order ID and an increasing version per order. Every event contains the complete order state. A given order ID and version must always describe the same content, including across separate deliveries. Deletes, partial updates, and concurrent writers are outside this example.
Prepare the source before MERGE
Delta Lake MERGE supports inserting new rows and updating existing ones. However, duplicate source rows can still be inserted when their key is absent from the target. Multiple source rows that attempt to update the same target row can also make the operation ambiguous. Prepare one source row per target key before merging. Databricks MERGE documentation
The preparation below removes exact duplicates, rejects conflicting versions within the batch, and selects the highest version for each order. The window assigns row numbers in descending version order. Apache Spark row_number documentation
Processing repeated and older order events
Figure 1. Prepare one candidate per order, then compare it with the target. Equal or older versions leave the target unchanged.
Implement the pattern in PySpark
Run this example in a Databricks Python notebook with Delta Lake support. Select a sandbox catalog and schema where you can create tables. The setup creates a new demonstration table and deliberately stops if that table already exists. To repeat the entire example, choose another unused table name. To test a replay, rerun only the calls to merge_orders.
```python
from delta.tables import DeltaTable
from pyspark.sql import functions as F, Window
TARGET = "replay_safe_orders_demo"
SCHEMA = "order_id LONG, source_version LONG, status STRING"
if spark.catalog.tableExists(TARGET):
raise ValueError("Choose an unused demo table name before setup.")
spark.sql(f"CREATE TABLE {TARGET} ({SCHEMA}) USING DELTA")
def merge_orders(batch_df):
source = batch_df.select("order_id", "source_version", "status")
invalid = source.filter(
F.col("order_id").isNull()
| F.col("source_version").isNull()
| F.col("status").isNull()
)
if invalid.limit(1).count():
raise ValueError("Required source fields cannot be null.")
unique = source.dropDuplicates()
conflicts = (
unique.groupBy("order_id", "source_version")
.count()
.filter(F.col("count") > 1)
)
if conflicts.limit(1).count():
raise ValueError("One order version has conflicting payloads.")
window = Window.partitionBy("order_id").orderBy(
F.col("source_version").desc()
)
latest = (
unique.withColumn("rn", F.row_number().over(window))
.filter(F.col("rn") == 1)
.drop("rn")
)
(
DeltaTable.forName(spark, TARGET).alias("t")
.merge(latest.alias("s"), "t.order_id = s.order_id")
.whenMatchedUpdateAll(
condition="s.source_version > t.source_version"
)
.whenNotMatchedInsertAll()
.execute()
)
batch = spark.createDataFrame([
(101, 1, "Created"),
(101, 2, "Shipped"),
(101, 2, "Shipped"),
(102, 1, "Created"),
], SCHEMA)
merge_orders(batch)
merge_orders(batch) # Replay the same delivery.
older = spark.createDataFrame([(101, 1, "Created")], SCHEMA)
merge_orders(older) # An older version must not replace Shipped.
actual = [tuple(row) for row in spark.table(TARGET)
.select("order_id", "source_version", "status")
.orderBy("order_id").collect()]
assert actual == [(101, 2, "Shipped"), (102, 1, "Created")]
```
The expected final result contains two rows. The assertion checks both the records and their values, so it detects an incorrect status as well as an unexpected duplicate. Collecting the result is appropriate for this tiny example; use distributed comparisons for large datasets.
Why the version condition matters
The match uses only order_id. A new key is inserted. For an existing key, the update condition permits only a strictly higher source_version. An equal version is a replay, and a lower version is an older delivery. Both leave the stored row unchanged. Conditional updates are part of the documented MERGE semantics. Databricks MERGE documentation
This design depends on the source contract. If the producer changes a payload while reusing a version, the example cannot reliably resolve that conflict across deliveries. Validate that contract upstream, or maintain an event history that allows those conflicts to be detected.
Applying the idea to streaming
For a custom Structured Streaming pipeline, a similar merge can run inside foreachBatch. Databricks documents at-least-once write guarantees for that callback, so its write logic must tolerate repeated execution. A checkpoint alone does not make arbitrary callback side effects idempotent. Databricks foreachBatch documentation
A complete streaming implementation also needs a durable checkpoint, empty-batch handling, and appropriate consumption of stateful query output. For CDC requirements involving deletes, history, or out-of-order changes, evaluate Lakeflow AUTO CDC rather than extending this small demonstration into a general CDC engine. Databricks CDC guidance
Tests to add before production
|Test |Expected behavior |
|----------------------------------------------------|-----------------------------|
|Repeat the same delivery |No change to current state |
|Repeat a row within a batch |One candidate for that order |
|Send an older version later |Keep the newer target version|
|Send a genuinely newer version |Update the order |
|Send conflicting payloads for one version in a batch|Reject the batch |
|Send a null key or version |Reject the batch |
Keep the target unique by order ID and coordinate its writers. This example starts with an empty table and one writer; it does not repair historical target duplicates or enforce uniqueness across independent writers. Retain raw deliveries separately when you need an audit trail, and monitor rejected records instead of silently discarding them.
The practical lesson is to define both identity and ordering before writing the merge. A business key answers which record an event belongs to. A source version answers whether that event should replace the state already stored.
How do you handle repeated deliveries and older updates in your Databricks pipelines? Do you use custom MERGE logic or Lakeflow AUTO CDC?
Announcements
