The Dup After Restart

Concepts covered: Structured Streaming

Tuesday morning, finance reports that 634 trips from overnight were paid twice. The payout query crashed at 02:13 and came back at 02:20. Duplicates after a restart have a short list of causes, and the checkpoint together with the duplicated rows tells you which one you are looking at.

The version of write_payout_lines running that night, before the rewrite in the previous section, appended each batch to the payouts database over JDBC and stamped every row with its batch id. Here is batch 41, step by step:

Each part behaved as designed. The files were still there, the checkpoint sent Spark back to batch 41 and nowhere else, and Spark produced the same 634 rows with the same batch id. The sink appended them because a plain append has no way to recognise a repeat. The half of the contract that was missing is the sink's.

Reading the duplicates

The duplicated rows carry the evidence. Group every duplicated trip by the batch id it was written under:

SELECT
  batch_id,
  COUNT(DISTINCT trip_id) AS duplicated_trips,
  COUNT(*) AS rows_written
FROM payout_lines
WHERE trip_id IN (
  SELECT
    trip_id
  FROM payout_lines
  GROUP BY trip_id
  HAVING COUNT(*) > 1
)
GROUP BY batch_id
ORDER BY batch_id

For this incident the query returns 1 row: batch 41, 634 duplicated trips, 1,268 rows written. Duplicates confined to a single batch id, 2 copies each, are the fingerprint of a replay into a sink that isn't idempotent. Other patterns point elsewhere:

Duplicates the source sends

The third row sits outside Spark's contract. Exactly-once means each input record lands once, and 2 deliveries of 1 event are 2 records. Phones cause this: an app that loses signal after sending an event, before the acknowledgement arrives, sends it again with the same event id. The app_event_deliveries table has that shape: 1 row per delivery, with its own delivery_id and received_ts, and event_uuid naming the event. Run the cell to compare deliveries with distinct events.

Any gap between the 2 numbers is redelivery. deliveries.dropDuplicates(["event_uuid"]) keeps 1 row per event. distinct() would keep every copy, because 2 deliveries of an event differ in delivery_id and received_ts. On a stream, the dedupe step must remember event ids from every earlier batch to catch a copy that arrives a day late, which makes it stateful; the last section of this lesson prices that memory.

After a restart, an analyst finds 9,400 duplicated trips. Every second copy carries a batch id between 0 and 12, while the first copies carry ids from 300 to 1,850. What happened?

• The checkpoint was replaced, so the query started over and re-read old files
• One batch re-ran after a crash, because the sink appends without checking ids
• The producer sent each trip twice, so Spark read 2 records for every trip
• The source's retention expired, so Spark skipped ahead to newer files

About this lesson

This section is part of the Structured Streaming: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.