What is foreachBatch?

Summary

• foreachBatch is a Structured Streaming sink that calls your function once per micro-batch with two arguments: a DataFrame holding that micro-batch's output and the batch's unique ID.
• It lets a stream reuse existing batch writers, write one micro-batch to several destinations and run operations that streaming DataFrames don't support, such as a Delta Lake MERGE.
• By default foreachBatch gives only at-least-once write guarantees. Use the batch ID to deduplicate or to make writes idempotent when you need exactly-once results.
• It works only with micro-batch execution. In continuous processing mode, foreach is the row-level alternative.
• Under Spark Connect the function runs on the server, can't use variables defined outside it, and should reach the session through the batch DataFrame's sparkSession.

What is foreachBatch?

foreachBatch is the bridge from a streaming query to ordinary batch code. You give a Structured Streaming query a function, and Spark calls it on the output of every micro-batch, passing the micro-batch as a regular DataFrame plus a unique batch ID. Inside the function you can run arbitrary batch logic: MERGE into a table, write to a system that has no streaming sink, or write the same rows to several places. The trade-off is that delivery semantics become your job, because Spark guarantees only at-least-once writes through it.

How foreachBatch works

foreachBatch(func) has been available in Scala, Java and Python since Spark 2.4; R isn't supported yet. Each call receives the output rows of one micro-batch as a DataFrame and the batch identifier. The PySpark docs add a useful guarantee: the output DataFrame is the same for the same batch ID, as long as every operation in the query is deterministic. The sink supports append, update and complete output modes, and whether it's fault-tolerant depends on what your function does.

from delta.tables import DeltaTable

def upsert(batch_df, batch_id):
    target = DeltaTable.forPath(batch_df.sparkSession, "/data/customers")
    (target.alias("t")
        .merge(batch_df.alias("s"), "s.id = t.id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute())

(spark.readStream.table("bronze.customer_updates")
    .writeStream
    .foreachBatch(upsert)
    .option("checkpointLocation", "/checkpoints/customers")
    .start())

What foreachBatch is for

Use	Why foreachBatch fits	Watch for
Writing to a system with no streaming sink	Reuses the existing batch writer on each micro-batch	Retries and duplicates are yours to handle
MERGE or upsert into a table	A micro-batch is a normal DataFrame, so batch-only operations work	Keep the merge idempotent; a restart can replay a batch
Writing one micro-batch to several places	You can write the batch DataFrame more than once	Each write can recompute the batch unless you persist it
Operations streaming DataFrames don't support	You apply them to each micro-batch instead	You reason about end-to-end semantics yourself

Delivery guarantees and idempotent writes

At-least-once means a failed or restarted query can run your function again for a batch that was already partly written. The Spark guide's answer is the batch ID: use it to deduplicate output and get an exactly-once guarantee. Delta Lake builds this into its writer with two options. txnAppId is a unique string for the application, and txnVersion is a monotonically increasing number such as the batch ID. Delta uses the pair to detect duplicate writes and ignore them. If you delete the checkpoint and restart with a new one, change txnAppId, because batch IDs start again from 0.

def write_twice(batch_df, batch_id):
    app_id = "customer-updates-fanout"
    batch_df.persist()
    for path in ["/data/sink_a", "/data/sink_b"]:
        (batch_df.write.format("delta").mode("append")
            .option("txnAppId", app_id)
            .option("txnVersion", batch_id)
            .save(path))
    batch_df.unpersist()

The persist() and unpersist() calls matter. Each write is an action, and without caching, each one can recompute the micro-batch, including re-reading input. With stateful queries, repeated actions also reload state within the same batch, which is why the guide recommends persisting inside the function.

foreachBatch under Spark Connect

foreachBatch has supported Spark Connect since Spark 3.5, but the PySpark docs warn that it behaves differently there. The function runs in a worker on the Spark Connect server, so it doesn't have access to variables defined outside it, and changes it makes to client-side variables never come back. Referencing the client's Spark session or a DataFrame defined outside the function raises STREAMING_CONNECT_SERIALIZATION_ERROR. Use batch_df.sparkSession instead, as the MERGE example above does.

How Databricks approaches foreachBatch

On Databricks, foreachBatch is the required path for Delta Lake merge operations in Structured Streaming. Handle empty DataFrames in your function, for example with if not batch_df.isEmpty():, because a Delta source can produce an empty micro-batch when an OPTIMIZE has no files to process. To write one stream to several destinations, Databricks recommends multiple Structured Streaming writers, since foreachBatch serializes those writes; if you do fan out to several Delta tables in one function, use the idempotent-write options. On standard access mode in Databricks Runtime 14.0 and above, print() output goes to the driver logs and everything the function references must be serializable. Lakeflow Pipelines offers a similar pattern as a ForEachBatch sink, declared with @dp.foreach_batch_sink() and fed by an append flow.

FAQs

Is foreachBatch exactly-once?

Not by default. It provides at-least-once writes, so deduplicate with the batch ID or use idempotent writes, such as Delta Lake's txnAppId and txnVersion options, to get exactly-once results.

What's the difference between foreach and foreachBatch?

foreach runs custom write logic on every row through open, process and close methods, and it works in continuous processing mode. foreachBatch hands you each micro-batch as a DataFrame and needs micro-batch execution.

Does foreachBatch work with the available-now trigger?

Yes. The available-now trigger (trigger(availableNow=True) in Python) processes all available data in one or more micro-batches and then stops, which is what foreachBatch needs. Continuous processing mode is excluded, and so is Real-time Mode (Spark 4.1 and later), whose supported sinks include foreach writers but not foreachBatch.