One key, one row: dedupe the source before you MERGE

The error message is the whole diagnosis: two rows from your source matched one row in the target, and MERGE refused to pick a winner. This is by design — by SQL semantics the update is ambiguous — so no retry, no WHEN MATCHED shuffle, and no bigger cluster fixes it. The fix the docs prescribe is preprocessing the source: shrink it to one row per key before the merge. For CDC and change feeds, that means keeping the latest change per key.

-- A CDC landing zone: two versions of customer 2 arrived together
CREATE OR REPLACE TEMP VIEW changes (id, first_name, last_name, salary, updated_at) AS
SELECT * FROM VALUES
  (2, 'Brian', 'Chen',   95000, TIMESTAMP '2026-09-25 09:01:00'),
  (2, 'Brian', 'Lee',    88000, TIMESTAMP '2026-09-25 09:00:00'),
  (4, 'Dana',  'Okafor', 70000, TIMESTAMP '2026-09-25 09:02:00');

-- Two source rows match target row id = 2: this fails, every time
MERGE INTO prod.customers AS t
USING changes AS s
  ON t.id = s.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;
-- [DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE]
-- Cannot perform Merge as multiple source rows matched and attempted
-- to modify the same target row in the Delta table in possibly conflicting ways.

The fix is one window function: number the changes per key by event time, keep row 1, then merge the clean source.

-- Keep the latest change per key, then merge the clean source
WITH deduped AS (
  SELECT * FROM (
    SELECT c.*,
           ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) AS rn
    FROM changes c
  )
  WHERE rn = 1
)
MERGE INTO prod.customers AS t
USING deduped AS s
  ON t.id = s.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

One caveat: the window orders by updated_at alone — if two changes for the same key can carry the same timestamp, the winner between them is arbitrary. Add a tiebreaker (a source sequence number or offset) to the ORDER BY to make the choice fully deterministic.

Recommended learning: Databricks docs: MERGE INTO

The failure

Illustrative scenario: A team ran a Kafka-to-Delta CDC pipeline: Debezium events into bronze, then a foreachBatch job that MERGEd each micro-batch into the customer dimension. After a broker hiccup, the producer retried and replayed a few hundred events. The staging table had a nightly dedupe job, so the team assumed the source was clean. But the replayed events landed inside a single micro-batch — and the merge source was the batch, not the table. The MERGE failed with DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE, the batch failed, and structured streaming retried the same poisoned batch on every trigger. The stream just stopped making progress.

1. Move the dedupe inside foreachBatch

Dedupe the batch DataFrame, not the table you cleaned last night. The duplicates that kill a merge are the ones in this batch.

from pyspark.sql import functions as F
from pyspark.sql.window import Window

def upsert_batch(batch_df, batch_id):
    # 1. Dedupe INSIDE the micro-batch: one row per key, latest event wins
    w = Window.partitionBy("customer_id").orderBy(F.desc("event_ts"))
    deduped = (batch_df
               .withColumn("rn", F.row_number().over(w))
               .filter("rn = 1")
               .drop("rn"))
    deduped.createOrReplaceTempView("batch_changes")

    # 2. Cache the merged source: MERGE reads it more than once,
    #    and caching keeps the input-rate metric honest
    deduped.cache()
    try:
        deduped.sparkSession.sql("""
          MERGE INTO prod.customers AS t
          USING batch_changes AS s
            ON t.customer_id = s.customer_id
          WHEN MATCHED AND s.event_ts > t.event_ts THEN UPDATE SET *
          WHEN NOT MATCHED THEN INSERT *
        """)
    finally:
        deduped.unpersist()

2. Make the merge idempotent

foreachBatch gives at-least-once guarantees: a restart can re-apply an already-merged batch. The s.event_ts > t.event_ts guard makes a replayed batch a no-op instead of a rollback — late or duplicate events can never move a row backward. The docs are explicit: verify that your merge statement inside foreachBatch is idempotent, otherwise restarts of the streaming query can apply the operation on the same batch of data multiple times.

3. Guard the grain, then validate

The dedupe must leave exactly one row per merge key: if the merge matches ON t.id = s.id alone but you partitioned by (id, region), two rows sharing an id across regions still collide on the same target row. Dedupe at the grain of the ON clause. Make this detector a habit — run it against every merge source and expect zero rows:

-- Expect zero rows: run against your merge source before every MERGE
SELECT id, COUNT(*) AS n
FROM changes
GROUP BY id
HAVING COUNT(*) > 1;

Running CDC merges you can't afford to get wrong? I run merge-safety audits — the duplicate-grain check, the idempotency review, and the replay drill, on your pipeline. Get in touch through the contact in About.

Recommended learning: Data engineering workflows — pipelines, orchestration, and streaming practice