Safe External Writes in Spark Structured Streaming with foreachBatch

When I first started using Spark Structured Streaming to write data to external systems like databases or APIs, I kept seeing duplicate records after job restarts. At first, I blamed the network or the sink connector. But after digging into checkpointing and micro-batch behavior, I realized the issue wasn’t the stream — it was how I was writing. That’s when foreachBatch became my go-to pattern for safe, reliable writes.

In Spark Structured Streaming, foreachBatch lets you process each micro-batch as a regular DataFrame or Dataset using standard batch operations. This is powerful because it gives you full control over the write logic — including transactions, idempotency checks, and exactly-once semantics — without giving up the benefits of streaming.

Here’s how I implement it in practice. First, I define a function that takes a batch DataFrame and its batch ID:

void writeBatch(DataFrame batchDF, long batchId) {
  if (batchDF.isEmpty()) return;

  // Use batchId for idempotency — e.g., store processed batch IDs in a deduplication table
  val processed = spark.sql("SELECT batch_id FROM streaming_metadata WHERE batch_id = $batchId")
  if (processed.count() > 0) {
    log.info("Batch $batchId already processed. Skipping.")
    return;
  }

  try {
    batchDF.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "target_table")
      .option("user", username)
      .option("password", password)
      .mode(SaveMode.Append)
      .save()

    // Mark batch as processed only after successful write
    spark.sql("INSERT INTO streaming_metadata (batch_id, processed_at) VALUES ($batchId, current_timestamp())")
  } catch {
    case e: Exception =>
      log.error("Failed to write batch $batchId", e)
      throw e  // Let Streaming handle retry via checkpoint
  }
}

Then I attach it to the stream:

streamingDF.writeStream
  .foreachBatch(writeBatch _)
  .outputMode("Update")
  .option("checkpointLocation", "/mnt/checkpoints/streaming_write")
  .start()

The key here is using the batchId — automatically provided by Structured Streaming — to detect and skip already-processed micro-batches. This makes the write idempotent. Combined with write-ahead logs and checkpointing, it delivers exactly-once guarantees even if the job fails mid-batch.

I’ve seen teams try to use foreach or mapPartitions for external writes, but those don’t give you batch-level control. Without batchId, you can’t safely track what’s been committed. I learned this the hard way after a Kafka connector glitch caused 20 minutes of duplicate invoices to hit our billing system.

A few practical tips:

• Always use a durable metadata store (like a database table or Delta Lake) to track processed batch IDs — not in-memory sets.
• Keep the idempotency check lightweight; avoid heavy queries that slow down the batch.
• If your external system supports transactions (e.g., PostgreSQL, SQL Server), wrap the write and metadata update in a single transaction.
• Never use SaveMode.Overwrite in foreachBatch unless you’re intentionally replacing a partition — it’s risky and defeats the purpose of streaming.

As I mentioned before in my post about monitoring systemd slices, observability is critical — so I always add metrics for batch processing time, success/failure counts, and lag. Pair that with alerting on stalled streams, and you’ve got a production-ready pipeline.

If you’re writing to systems that don’t natively support exactly-once (like REST APIs or legacy DBs), foreachBatch with idempotency keys is often the most practical path forward. It’s not magic — it’s just disciplined batch processing inside a streaming framework. And in my environment, that’s what keeps the data accurate and the alerts silent.

Cover image: learn_tek · CC0 (Openverse / kamu malı) · https://www.flickr.com/photos/153724200@N07/27774351928