Exactly-once in streaming: practical tips and trade-offs

Introduction

Streaming systems promise low-latency insights, but without proper safeguards, they deliver duplicates and lost records instead. Exactly-once in streaming: practical tips and trade-offs requires balancing idempotency, checkpointing, and sink semantics to guarantee each event is processed and persisted exactly once. In this article, we'll walk through battle-tested patterns from production e-commerce and SaaS pipelines that prevent duplicates from reaching your warehouse while keeping recovery fast.

Checkpoints track progress

Checkpointing is the foundation of recovery in Spark Streaming. When your job crashes, the engine restarts from the last committed checkpoint, re-reading only the messages it hasn't processed yet. Without checkpoints, you either replay everything (duplicates) or skip data (loss). The checkpoint directory stores offset ranges, shuffle state, and idempotent write identifiers that Spark uses to resume safely.

In structured streaming, you enable checkpointing by specifying a path in your write operation. That path must be on a reliable filesystem—ideally a cloud object store with strong consistency semantics like S3 with eventual consistency handled by the checkpoint protocol, or better yet, HDFS or Azure Data Lake.

query = (df
    .writeStream
    .format("delta")
    .mode("append")
    .option("checkpointLocation", "/mnt/checkpoints/orders_stream")
    .table("orders_target"))

query.awaitTermination()

The checkpoint directory contains a metadata file with the current offset and UUID for idempotent writes. Never share checkpoint locations between different jobs—each query needs its own. We've seen teams reuse checkpoint paths and lose track of which offsets belong to which query, leading to either duplicate processing or silent data loss.

In production, monitor checkpoint latency. If checkpointing takes longer than your micro-batch interval, you're in trouble: the next batch starts before the previous one is safely persisted, so a failure mid-batch leaves you vulnerable. We typically aim for checkpoint time under 10% of batch duration.

Idempotent sinks like Delta

An idempotent sink is one where writing the same record twice produces the same result as writing it once. Delta Lake achieves this through its transaction log and unique write identifiers. When Spark writes with Delta, it attaches a UUID tied to the checkpoint; if the job crashes and restarts, it uses the same UUID for the same batch, and Delta deduplicates on write.

Delta is the gold standard for streaming sinks in the Lakehouse because it supports both idempotent appends and ACID semantics for merges. Let's compare sinks by their exactly-once guarantees:

For a simple append-only workload, Delta is your answer. Your stream reads from Kafka, processes events, and appends to a Delta table:

df = (spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "events")
    .load())

parsed = df.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

query = (parsed
    .writeStream
    .format("delta")
    .mode("append")
    .option("checkpointLocation", "/mnt/ck/events")
    .table("raw_events")
    .awaitTermination())

The Delta write is idempotent: if the batch committed to Kafka offset 1000 but the checkpoint wasn't written, restart will reprocess messages up to 1000, and Delta will silently drop the duplicates thanks to its write UUID.

foreachBatch with MERGE

Sometimes you can't simply append. You need to update existing records—for instance, when processing user profile updates or inventory changes. Delta's MERGE statement combined with foreachBatch lets you upsert with exactly-once guarantees.

foreachBatch is a sink that gives you a micro-batch DataFrame at each interval. You control how it's written; you can call a MERGE, invoke an HTTP endpoint, or write to a database. The key is that foreachBatch is idempotent: if your code crashes during the write, the entire batch is replayed from the source, so you can safely use MERGE without deduplication logic.

def merge_updates(batch_df, batch_id):
    batch_df.write.format("delta").mode("overwrite").option(
        "replaceWhere", 
        f"batch_id = {batch_id}"
    ).save("/mnt/target/table")

df = (spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "updates")
    .load())

parsed = df.select(from_json(...).alias("data")).select("data.*")

query = (parsed
    .writeStream
    .foreachBatch(merge_updates)
    .option("checkpointLocation", "/mnt/ck/updates")
    .start())

A more robust pattern uses MERGE to handle both inserts and updates. Here's an e-commerce example: you're streaming order events, and need to update the order status in your warehouse. MERGE ensures that even if you replay a batch, the final state is correct.

def merge_orders(batch_df, batch_id):
    batch_df.createOrReplaceTempView("updates")
    
    spark.sql("""
        MERGE INTO orders_target t
        USING updates s
        ON t.order_id = s.order_id
        WHEN MATCHED AND s.timestamp > t.last_update THEN
            UPDATE SET 
                status = s.status,
                last_update = s.timestamp
        WHEN NOT MATCHED THEN
            INSERT (order_id, status, last_update)
            VALUES (s.order_id, s.status, s.timestamp)
    """)

query = (parsed
    .writeStream
    .foreachBatch(merge_orders)
    .option("checkpointLocation", "/mnt/ck/orders_merge")
    .start())
    
query.awaitTermination()

The critical detail: the foreachBatch function must be idempotent. If you pass batch_id to the function, you can store it in your table and deduplicate on replay. The combination of batch_id and your natural key (order_id) ensures that replaying a batch doesn't double-update.

Test restarts and failures

Theory is one thing; practice is another. We recommend building chaos tests into your streaming pipeline CI/CD. Kill the driver mid-batch, unplug the network, or corrupt the checkpoint file and verify recovery behavior. Most duplicate and loss bugs only surface under failure conditions.

A practical testing approach for Spark Streaming:

1. Start a local Spark job with a small batch interval (e.g., 5 seconds).
2. Source a Kafka topic with a known number of records.
3. After a few batches, send a SIGTERM to the driver.
4. Restart the job and let it run to completion.
5. Verify record counts and deduplication in the target table.

def test_streaming_restart():
    """Test that restarting doesn't cause duplicates."""
    spark = SparkSession.builder.appName("restart_test").getOrCreate()
    
    # Write test records to Kafka
    test_records = [{"id": i, "value": f"record_{i}"} for i in range(1000)]
    df_test = spark.createDataFrame(test_records)
    df_test.write.format("kafka").option("kafka.bootstrap.servers", "kafka:9092")\
        .option("topic", "test_restart").save()
    
    # Run streaming job
    query = (spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("subscribe", "test_restart")
        .load()
        .writeStream
        .format("delta")
        .option("checkpointLocation", "/tmp/test_ck")
        .table("test_target")
        .start())
    
    # Let it run, then stop mid-stream
    time.sleep(10)
    query.stop()
    
    # Restart
    query = (spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("subscribe", "test_restart")
        .load()
        .writeStream
        .format("delta")
        .option("checkpointLocation", "/tmp/test_ck")
        .table("test_target")
        .start())
    
    query.awaitTermination()
    
    # Verify
    result = spark.sql("SELECT COUNT(DISTINCT id) FROM test_target")
    assert result.collect()[0][0] == 1000, "Duplicates or data loss detected"

Run these tests in your staging environment on real data volumes. We've caught bugs where checkpoint corruption or clock skew in a multi-node cluster led to silent failures that only appeared under stress.

Common mistakes

• Sharing checkpoint locations: Two queries writing to different tables but using the same checkpoint directory will corrupt offsets and cause duplicates. Each streaming job needs a unique checkpoint path.
• Checkpoints on ephemeral storage: If your checkpoint lands on a local SSD that gets recycled on node restart, you lose recovery guarantees. Always use distributed, persistent storage.
• Not testing failure scenarios: A job that runs for hours without failure can hide exactly-once bugs. Force crashes in staging to validate recovery.
• Relying on sink retries for idempotency: A sink that retries on network error doesn't guarantee exactly-once; you still need idempotent writes or deduplication logic in the sink itself.
• Forgetting to handle late or out-of-order data: Even with exactly-once writes, late-arriving events can produce logical inconsistencies. Use watermarks and versioning to manage event ordering.

Key takeaways

• Checkpoints are your safety net: they store offset ranges and write UUIDs so restarts can resume without duplicates.
• Delta Lake provides idempotent appends out of the box; use it for simple append-only streams.
• foreachBatch with MERGE enables upserts while maintaining exactly-once semantics; always include batch_id for deduplication.
• Test failure scenarios in staging: kill drivers, pause sources, and verify that record counts stay correct.
• Avoid shared checkpoints, ephemeral storage, and sink-only retries; these are the most common sources of duplicates and data loss in production.