Safe External Writes with foreachBatch in Spark Structured Streaming

In my environment, Spark Structured Streaming is the go-to for real-time data pipelines, but writing safely to external systems like databases or APIs remains one of the trickiest parts. The core challenge is ensuring data consistency when failures happen — you don’t want duplicates, and you can’t afford lost data. That’s where foreachBatch combined with proper checkpointing and idempotent writes becomes essential.

Why foreachBatch Over foreach

When I first started with Structured Streaming, I tried using foreach for external writes. It seemed simple: iterate over each row and call an external API. But this approach breaks exactly-once semantics because there’s no transactional boundary around the write operation. If a task fails after writing but before committing, you get duplicates on restart.

foreachBatch solves this by giving you a micro-batch DataFrame and a batch ID. You get a clear boundary: process the entire batch, then commit or roll back as a unit. This makes it easier to implement idempotent logic — for example, using a unique batch ID to skip already-processed data in your external system.

Implementing Idempotent Writes

Idempotency means writing the same data multiple times has the same effect as writing it once. In practice, I do this by including the Spark batch ID in the write operation. For example, when writing to a PostgreSQL table, I add a batch_id column and use it in an UPSERT:

INSERT INTO events (event_id, payload, batch_id)
VALUES (?, ?, ?)
ON CONFLICT (event_id, batch_id) DO UPDATE SET payload = EXCLUDED.payload;

This ensures that if the same batch is reprocessed due to a failure, the write has no net effect. I avoid using timestamps or row-level IDs alone because they can’t guarantee idempotency across retries.

Checkpointing for Recovery

Checkpointing is non-negotiable. Without it, Spark can’t recover the state of your streaming query after a failure. I always configure a reliable checkpoint location — preferably on a distributed filesystem like HDFS or S3:

query = df.writeStream \
    .foreachBatch(write_to_external) \
    .outputMode("update") \
    .option("checkpointLocation", "s3://my-bucket/checkpoints/events-pipeline") \
    .start()

The checkpoint stores not just the metadata but also the progress of each micro-batch. If the driver fails, Spark can resume from the last successfully committed batch, ensuring no data is lost or duplicated.

Handling Failures Gracefully

Even with idempotent writes and checkpointing, you need to handle transient failures. I wrap the external write in a retry loop with exponential backoff, but only for network-level issues — never for data validation errors. For example:

def write_to_external(batch_df, batch_id):
    try:
        # Convert to Pandas or use JDBC for batch write
        batch_df.write \
            .format("jdbc") \
            .option("url", jdbc_url) \
            .option("dbtable", "events") \
            .option("user", db_user) \
            .option("password", db_password) \
            .mode("append") \
            .save()
    except Exception as e:
        if "network" in str(e).lower() or "timeout" in str(e).lower():
            raise  # Let Spark handle retry via task retry
        else:
            # Log and skip invalid data — don’t fail the batch
            log_error(batch_id, e)
            batch_df.filter("false").write.format("noop").save()  # Ack empty batch

This way, transient issues trigger Spark’s built-in retry mechanism, while bad data is isolated without killing the whole batch.

Monitoring and Validation

I monitor lag and throughput using Spark’s streaming metrics, but I also add end-to-end validation. In my setup, I write a checksum of each batch to a separate table and compare it with the source during quiet periods. If there’s a mismatch, I trigger an alert — this has caught silent failures that metrics alone missed.

Conclusion

Using foreachBatch with idempotent writes and proper checkpointing isn’t just a best practice — it’s the foundation of reliable external writes in Spark Structured Streaming. It gives you exactly-once semantics without relying on external transaction systems, which is crucial when working with SaaS APIs or eventually consistent stores. As I mentioned before in my post about latency SLOs in Argo Rollouts, consistency in the data layer directly impacts the reliability of downstream systems — so invest the time to get this right.

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