Pranaav Srinivasan

All posts

Why a streaming pipeline needs an idempotent sink

Kafka delivers at least once, so duplicates are normal. Here is how deduplication in the stream and an idempotent write at the end keep a pipeline correct.

· 2 min read · Kafka, Spark, Snowflake

Duplicates in a streaming pipeline are not a bug. They are what you get when you choose not to lose data. Kafka's default guarantee is at least once: if something fails midway, the safe move is to try again, and trying again can deliver the same record twice.

The question is where you deal with them. The answer I keep coming back to is two places: once inside the stream, and once at the very end, in the write.

Where duplicates come from

  • Producer retries. A producer sends a record, the acknowledgement is lost, and it sends the record again. Enabling the idempotent producer (enable.idempotence=true) removes this one within a partition and a producer session.
  • Consumer restarts. A job crashes after processing a batch but before the offsets are committed. On restart it reads that batch again.
  • Upstream resends. The source system emits the same business event twice, with different Kafka offsets. Nothing in Kafka can see this one.

The first is a transport problem and the platform can fix it. The other two are application problems, and only your pipeline knows what “the same event” means.

Step one: deduplicate in the stream

Give every event a stable identifier at the source, then drop repeats in Structured Streaming. A watermark bounds how long Spark remembers identifiers, which keeps state from growing forever.

from pyspark.sql.functions import col, from_json

events = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", BROKERS)
    .option("subscribe", "payments")
    .load()
    .select(from_json(col("value").cast("string"), schema).alias("e"))
    .select("e.*")
)

deduped = (
    events
    .withWatermark("event_time", "10 minutes")
    .dropDuplicatesWithinWatermark(["event_id"])
)

dropDuplicatesWithinWatermark is available from Spark 3.5. On older versions, include event_time in the column list passed to dropDuplicates so the watermark can expire old state.

The watermark is a trade-off you choose on purpose. A duplicate that arrives after the window is no longer caught here. A longer window catches more and costs more memory.

Step two: make the write idempotent

Deduplication in the stream cannot cover everything. A micro-batch can be written, the job can fail before the checkpoint is saved, and the same batch is written again. The defence is a sink where writing the same record twice has the same effect as writing it once.

In a warehouse, that usually means writing each micro-batch to a staging table and merging on the event identifier:

MERGE INTO curated.payments AS t
USING staging.payments_batch AS s
  ON t.event_id = s.event_id
WHEN NOT MATCHED THEN
  INSERT (event_id, event_time, amount, account_id)
  VALUES (s.event_id, s.event_time, s.amount, s.account_id);

Replaying a batch now inserts nothing the second time. In Spark, foreachBatch hands you the batch and a batch_id, which is the natural place to run this merge.

A short checklist

  • Does every event carry an identifier that survives retries and resends?
  • Is the watermark long enough for the latest duplicate you have actually seen?
  • Can the sink accept the same batch twice without changing the result?
  • Do you monitor the duplicate rate, so a spike is visible before it matters?

Exactly-once is not a switch in Kafka or in Spark. It is a property of the whole pipeline, and it holds only when each stage does its part.