Real ingestion produces duplicates: a source retries, a file gets reprocessed, an event fires twice. Removing them looks trivial — "remove duplicates" — until you notice the word duplicate hides two decisions, and that the one-line fix that works on a batch quietly destroys a streaming job. This lesson makes both precise.

The spine

Beat 1 — the anchor: two questions before any dedup

Anchor. Before removing duplicates, answer two things. (1) Duplicate by what? — identical whole rows, or "same key, keep one specific survivor"? (2) Is the data bounded (a finite batch you can see all at once) or unbounded (a stream, where "have I seen this before?" needs remembered state)? Every technique is just an answer to those two.

Miss question 2 and you write a dedup that needs infinite memory. That's the trap this lesson exists for.

Beat 2 — batch: identity vs keep-latest

Identical rows (easy). The whole dataset is in view, so Spark compares every row:

• SELECT DISTINCT * FROM t / df.dropDuplicates() — remove fully identical rows.
• df.dropDuplicates(["user_id", "event_date"]) — dup by a subset of columns: one row per key, but arbitrary which one.

Predict: that "arbitrary which one" — when is it not what you want?

…

Almost always. You usually want a specific survivor — the latest, the most complete — and dropDuplicates(keys) can't express that. That's where last lesson pays off:

SELECT * FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_ts DESC) AS rn
  FROM events
) WHERE rn = 1;   -- one row per user_id: the most recent

PARTITION BY the key, ORDER BY newest-first, keep ROW_NUMBER() = 1. ROW_NUMBER specifically — it's unique per partition, so exactly one survivor (RANK/DENSE_RANK would keep ties, from Advanced transformations — window functions, joins, aggregations).

Beat 3 — streaming: the state trap, and why a watermark is mandatory

Now run the same dedup as a stream.

Predict: dropDuplicates(["id"]) on a stream must recognise a repeat of a row seen an hour ago. What must it remember — and what happens to that memory on an endless stream?

…

It must remember every key it has ever seen — that "seen" set is state (Structured Streaming & the state model), held in the state store, and on an unbounded stream it grows forever until the job's memory dies.

The fix is the same tool that bounded streaming aggregations — a watermark: "records can't be later than N behind the max event time, so anything older can be forgotten." That caps dedup state to a time window:

• df.withWatermark("event_ts", "1 hour").dropDuplicates(["id"]) — keeps the "seen" state ~1 hour back. (Docs warn its state can still grow when duplicates carry different event times.)
• dropDuplicatesWithinWatermark(["id"]) (watermark set) — purpose-built: dedups within the window even when duplicates' event times differ, and bounds state cleanly. The current recommended streaming-dedup function (DBR 13.3 LTS+).

Lock it. Batch keep-latest = ROW_NUMBER()…=1. Streaming dedup with no watermark = unbounded, ever-growing state. "Streaming + dedup" ⇒ watermark — an answer that omits it is wrong.

The dials (skim now; return when a question needs one)

◆ The streaming pattern the exam favours: dedup-in-batch + insert-only merge

The recommended streaming-dedup shape isn't just dropDuplicates — it's de-duplicate within each micro-batch, then insert-only MERGE into the target:

MERGE INTO orders t USING microbatch s
  ON t.order_id = s.order_id AND t.order_timestamp = s.order_timestamp
WHEN NOT MATCHED THEN INSERT *   -- existing keys skipped → idempotent

Only WHEN NOT MATCHED, so an already-present key is a no-op and re-running inserts nothing new — the ingest is idempotent (run twice, same result). Inside foreachBatch the merge must be idempotent, or a restart replays the batch.

◆ The batch trap: dropDuplicates dedups the batch, not the target

A favourite question: a nightly job does dropDuplicates then appends. Each write contains unique records — but records may duplicate rows already in the target from earlier writes. dropDuplicates only sees the current batch. To dedup against history you need the insert-only merge (which checks the target), often on a natural composite key (e.g. user_id, review_id, product_id, review_timestamp). Likewise a watermark that keeps only ~2h of state means duplicates arriving more than ~2h apart may be retained.

◆ Write-level idempotency: txnAppId + txnVersion

For retry-safety across failures, Delta accepts txnAppId + txnVersion on the write — it records them in the log and silently ignores a re-applied (appId, version), so a failed-then-retried batch can't double-write.

◆ The MERGE limitation that forces dedup

MERGE errors when multiple source rows match the same target row — the outcome would be ambiguous (which source row wins?), so Delta refuses. The fix is the keep-latest idiom: dedup the source first so each target key matches ≤1 source row. That's why dedup and merge travel together. (Aside: with schema evolution, a MERGE adding a new column reads existing rows as NULL for it — no rewrite, per How Delta Lake works — the transaction log.)

◆ Name the collision: dropDuplicates vs MERGE

dropDuplicates dedups on identity/columns, keeps an arbitrary row; MERGE dedups on a key, lets you control the survivor (or skip existing). "Remove exact repeat rows" → dropDuplicates/DISTINCT; "one row per business key, upsert-style, idempotent" → MERGE. Change feed with ordering + SCD? Don't hand-write it — AUTO CDC (APPLY CHANGES — CDC and SCD, declaratively) does dedup + ordering + SCD for you.

Takeaways (rebuild it from these)

1. Define the duplicate first: by identity (DISTINCT/dropDuplicates) or by key (keep a specific survivor).
2. Keep-latest per key = ROW_NUMBER() OVER (PARTITION BY key ORDER BY ts DESC) then = 1 — ROW_NUMBER because it's unique per partition.
3. Streaming dedup needs a watermark — plain dropDuplicates on a stream = unbounded state. Prefer dropDuplicatesWithinWatermark. No watermark ⇒ wrong.
4. Recommended streaming shape: dedup-in-batch + insert-only MERGE (idempotent; foreachBatch merge must be idempotent). dropDuplicates dedups the batch, not the target — use the merge for cross-history. txnAppId/txnVersion = retry-safe writes.
5. MERGE errors on multiple source rows matching one target → dedup source first. Collision: dropDuplicates (identity, arbitrary) vs MERGE (key, chosen). Change feeds → AUTO CDC.

Before you move on — say these without scrolling up

1. The two questions to answer before any dedup.
2. Keep-latest per key — which window function exactly, and why not RANK?
3. Streaming dropDuplicates with no watermark — what grows, and what's the fix?
4. A nightly dropDuplicates-then-append still leaves dupes in the target — why, and what fixes it?

Next: once deduped, we enforce that the data is correct — expectations and constraints, and what each does to a bad row → Data quality — expectations and constraints, and what happens to a bad row.