Designing Exactly-Once Spark Structured Streaming Pipelines That Survive Failures

The requirement: no loss, no duplicates, even when things die

Real-time ETL pipelines built on Apache Spark Structured Streaming usually fail in one of two ways: records disappear after a crash, or the same records land in the sink twice. Both are unacceptable when the downstream system is a billing table, a regulatory report, or a feature store. The useful takeaway: Structured Streaming can deliver end-to-end exactly-once semantics, but only if three conditions hold simultaneously — a replayable source, a reliable checkpoint directory, and an idempotent or transactional sink. If any one of those is missing, you have at-least-once with duplicates, whether or not your dashboards show it.

This note walks through the smallest design that meets the requirement, the trust boundaries, how to verify it actually works, and the conditions under which you should pick a different architecture. Version assumption: behavior described matches Spark 3.x (3.3–3.5); verify against your release's Structured Streaming documentation before production rollout.

The smallest suitable design

The minimal exactly-once pipeline has four components:

1. A replayable source. Apache Kafka is the common choice. Spark tracks offsets itself rather than committing them back to Kafka, so a restarted query re-reads exactly the data it had not yet committed.
2. A checkpoint directory on durable storage. HDFS or S3. The checkpoint holds two things: committed source offsets and, for stateful operations, the state store. This is the write-ahead log that makes recovery possible.
3. A micro-batch query. Each trigger interval, Spark reads a batch, processes it, writes the sink, then commits the offset range to the checkpoint. The commit happens only after the sink write succeeds.
4. An idempotent sink. Either a sink that natively supports transactions (Delta Lake, some JDBC targets with foreachBatch plus a transaction) or one where re-writing the same batch is harmless (overwrite by deterministic key, upsert by primary key).

A representative skeleton:

val query = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092")
  .option("subscribe", "events")
  .option("failOnDataLoss", "false") // see failure modes below
  .load()
  .selectExpr("CAST(value AS STRING) as payload")
  .writeStream
  .format("delta")
  .option("checkpointLocation", "s3a://my-bucket/checkpoints/events-etl")
  .outputMode("append")
  .trigger(org.apache.spark.sql.streaming.Trigger.ProcessingTime("1 minute"))
  .start()

Run this from your Spark application entry point (e.g., via spark-submit on the cluster); it requires an IAM role or Hadoop credentials with write access to the checkpoint bucket and network access to the Kafka brokers. The Delta sink is chosen here because it supports transactional writes per micro-batch; swapping in a plain Parquet file sink breaks exactly-once because a failed batch can leave partial files.

Trust and data boundaries

Draw the boundary carefully, because exactly-once is only as strong as its weakest edge:

• Source boundary: Kafka must retain data long enough for your worst-case downtime. If retention is 24 hours and the job is down for 36, offsets in the checkpoint point at data that no longer exists. That is data loss the framework cannot prevent.
• Internal boundary: Everything between the checkpoint and Spark's execution engine is trustworthy — Spark replays from the last committed offset and rebuilds state from the checkpointed state store.
• Sink boundary: Trust ends at the sink. Spark can guarantee it will attempt each batch's write exactly once after recovery, but if the sink cannot deduplicate or roll back a partial write, duplicates appear. A JDBC sink writing plain INSERTs is the classic violation; wrap it in foreachBatch with a transaction keyed on batch ID, or use a merge/upsert.

Operational checks

Exactly-once correctness is silent when it works, so monitor the pipeline's health instead:

• Consumer lag. Attach a StreamingQueryListener and compare inputRowsPerSecond against processedRowsPerSecond from each progress event. A sustained gap means the stream is falling behind and will eventually hit source retention limits.
• Batch duration vs. trigger interval. If durationMs for the whole batch regularly exceeds the trigger interval, you are accumulating backlog. Either scale executors or lengthen the trigger.
• Checkpoint growth. Checkpoint directories accumulate state-store versions and offset logs. Spark retains a bounded number of state versions by default, but compacted metadata and old logs still grow over long uptimes; schedule periodic inspection of the directory size, since an enormous checkpoint slows recovery and can degrade filesystem metadata performance.
• Small files. Aggressive triggers (e.g., every few seconds) on file-based sinks produce many tiny files, which hurts downstream read performance. Prefer longer triggers and periodic compaction (Delta's OPTIMIZE) over very low latency if the sink is columnar files.

Failure modes and what actually happens

Driver failure: The driver holds the offset log. On restart with the same checkpoint location, the query resumes from the last committed offset. Batches that were in flight are re-executed — which is exactly why the sink must be idempotent. Expect a recovery delay while the driver reloads state; large state stores make this slower.

Executor failure: The cluster manager replaces the executor and Spark retries the affected tasks. No offset rollback is needed for stateless work; stateful tasks reload their partition's state from the checkpointed state store.

Source data loss: With Kafka, failOnDataLoss defaults to true and the query fails loudly if checkpointed offsets have aged out of retention. Setting it to false (as in the example, sometimes used during development) converts that loud failure into silent data skipping — a deliberate trade-off you should not carry into production.

Checkpoint corruption or deletion: Unrecoverable for the running query's semantics. You must start a new query with a fresh checkpoint and manually choose starting offsets. Treat the checkpoint directory as production data: restrict delete permissions and never share one checkpoint between two queries.

Verifying the guarantee before you trust it

Do not assume exactly-once works because the documentation says so — test the failure path in a staging environment:

1. Deploy the job with a Kafka source and a checkpoint on your real storage layer.
2. Produce a known sequence of records with unique IDs (e.g., 100,000 sequential integers).
3. While the stream is actively processing, kill the driver process (kill -9 on the driver host, or delete the pod in Kubernetes). Restart the job against the same checkpoint.
4. Query the sink: count records, count distinct IDs. The counts must be equal, and the total must eventually equal the number produced. Any gap or duplicate means one of the three preconditions — replayable source, durable checkpoint, idempotent sink — is not actually holding.

Repeat with an executor kill instead of a driver kill. This is a manual verification procedure; results depend on your sink and configuration, so run it against your own stack rather than trusting a generic claim.

Conditions that should change the design

• Latency below the micro-batch interval. If you need sub-second end-to-end latency, micro-batching is the wrong tool. Spark's Continuous Processing mode exists but supports a restricted set of operations and only at-least-once semantics in practice; evaluate a true record-at-a-time engine (Flink, Kafka Streams) instead.
• Non-idempotent sink you cannot change. If the sink is a third-party API with no deduplication keys, no checkpointing strategy will save you. Insert a transactional staging layer (e.g., Delta table) and reconcile downstream separately.
• Very large state with strict recovery-time targets. Recovery time scales with state size. If restart time matters, consider RocksDB state store with changelog checkpointing (available in Spark 3.2+) to shorten recovery, or partition the workload into smaller queries.

The design is deliberately boring: replayable source, durable checkpoint, idempotent sink, monitored lag. Every deviation from that list is a conscious trade against the exactly-once guarantee — make sure the trade is written down before production makes it for you.