Deduplicate the Stream

A hard Spark interview practice problem on DataDriven. Write the Spark code and run it against the tables, with instant feedback.

Domain
Spark
Difficulty
hard
Seniority
L5
Asked in
Databricks

Problem

Since the broker restart, our consumers keep receiving the same event again under a fresh msg_id, so the only identity that survives a replay is the producer's key in part_key, and a message that arrived without one still counts as a delivery but matches no event. From stream_msgs, show each named consumer's total deliveries next to how many unique events it actually received.

Kafka replayed 15 minutes. Your dashboard doubled.

Expected output

First 5 of 7 rows
consumerdeliveriesunique_events
analytics-svc205
data-warehouse205
fraud-detector205
ml-pipeline205
notification-svc205

Worked solution and explanation

This is replay auditing dressed up as a dashboard fix. Kafka delivers at least once. So the delivery count is what the broker handed you, the event count is what exactly-once output should have been, and the gap between them IS the inflation. Anyone can count rows. The trick is choosing the identity. msg_id is minted fresh on every redelivery, so a unique count of it counts every replay as new. Key on the producer's part_key instead, or the audit reports zero duplicates on a stream that doubled the dashboard.

Two counts, one replay

Step 1: Keep only named consumers

A message with a NULL consumer was never routed anywhere, so it belongs in no one's audit. Filter it out before the groupBy. Otherwise it surfaces as its own ghost row in the result.

Step 2: Count every delivery

F.count("*") per consumer is the raw at-least-once picture: every redelivery and every unkeyed message. This is the number the inflated dashboard was effectively summing.

Step 3: Count each event key once

F.countDistinct("part_key") collapses replays of the same event to one. It also skips NULL keys, which is exactly the ruling. An unkeyed message is a delivery with no event behind it, so it lands in the first column and not the second.

Deliveries vs unique events per consumer
routed = spark.table("stream_msgs").filter(F.col("consumer").isNotNull())

result = (
    routed
    .groupBy("consumer")
    .agg(
        F.count("*").alias("deliveries"),
        F.countDistinct("part_key").alias("unique_events"),
    )
)

Both counts come out of the same aggregate on consumer.

Counts deliveries
F.count("*") counts each redelivery and each unkeyed message. It answers what the broker sent.
Counts events
F.countDistinct("part_key") counts each producer key once and ignores NULL. It answers what actually happened.

Back in the live stream

The batch audit tells you how bad the replay was. The streaming job has to stop it from happening. There, the same stable key goes into dropDuplicatesWithinWatermark, and the watermark keeps the state store from remembering every key forever. A replay that spans several 2-minute triggers is still caught, because the state outlives any single micro-batch.

The streaming fix (fragment, not the graded answer)
deduped = (
    clicks
    .withWatermark("event_time", "30 minutes")
    .dropDuplicatesWithinWatermark(["click_id"])
)

State for a key expires once the watermark passes it, so executor memory stays bounded.

Common follow-up questions