How do I write a stream to multiple sinks?

Summary

• Structured Streaming documents two options: start one streaming query per sink, or run one query whose foreachBatch function writes each micro-batch to every destination.
• Give every query its own checkpoint location. Each query records its own progress there and recovers on its own.
• In foreachBatch, persist the batch before the writes and unpersist it afterwards, so Spark doesn't recompute the output for each write.
• foreachBatch is at-least-once by default. Deduplicate with the batch ID, or use Delta Lake's txnAppId and txnVersion options for idempotent Delta writes.
• Databricks recommends a separate streaming write per sink, because foreachBatch serializes the writes.

How do I write a stream to multiple sinks?

Apache Spark Structured Streaming documents two ways to send one stream to several sinks. You can start one streaming query per sink, each with its own checkpoint location, or run a single query whose foreachBatch function writes every micro-batch to each destination. Separate queries keep the sinks independent. foreachBatch lets you reuse batch writers and send the same micro-batch to several places, but you manage recomputation and idempotency yourself.

How do I run one streaming query per sink?

The guide says "You can start any number of queries in a single SparkSession", and they run concurrently, sharing the cluster's resources. Define the source once, then start a query for each sink.

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

to_delta = (events.writeStream.format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/chk/events_delta")
    .toTable("events"))

to_files = (events.writeStream.format("parquet")
    .option("checkpointLocation", "/chk/events_files")
    .option("path", "/data/events_files")
    .start())

spark.streams.awaitAnyTermination()

Each query saves its progress, meaning the offsets processed in each trigger plus any running aggregates, to its own checkpoint location. Each query also reads the source separately. With Kafka, each query generates a unique consumer group ID by default, so every query can read all partitions of its subscribed topics.

How do I write to several sinks from one query?

foreachBatch calls your function with two parameters: a DataFrame holding the micro-batch output and the micro-batch's unique ID. Writing that DataFrame several times can recompute it, including re-reading the input, so the guide says to persist the batch, write it to each location, then unpersist it.

app_id = "events-fanout"  # unique string; change it if you reset the checkpoint

def write_to_sinks(batch_df, batch_id):
    if batch_df.isEmpty():
        return
    batch_df.persist()
    for path in ["/data/events_main", "/data/events_audit"]:
        (batch_df.write.format("delta")
            .option("txnVersion", batch_id)
            .option("txnAppId", app_id)
            .mode("append")
            .save(path))
    batch_df.unpersist()

(events.writeStream
    .foreachBatch(write_to_sinks)
    .option("checkpointLocation", "/chk/events_fanout")
    .start())

Delta Lake uses the txnAppId and txnVersion pair to spot duplicate writes and skip them when a failed batch reruns. The options need Delta Lake 2.0.0 or above. If you delete the checkpoint and start a new one, pass a different txnAppId, because batch IDs restart from 0 and Delta would ignore the new writes. For sinks without such options, use batch_id to deduplicate the output yourself.
Two more limits from the guide: foreachBatch doesn't work in continuous processing mode, and in a stateful query each extra action on the batch DataFrame reloads state unless you persist it.

Which approach should I choose?

Option	When to use it	Trade-off
One query per sink	Every destination has a streaming sink and can progress on its own	Each query reads the source and keeps its own checkpoint
One query with foreachBatch	The same micro-batch must reach several places, or a sink only has a batch writer	At-least-once by default; persist the batch and make every write idempotent
foreach	Continuous processing mode, or no batch writer exists	You write row-level open, process and close logic

What delivery guarantees do I get?

It depends on the sink. Spark's sink table lists the file sink as exactly-once, the Kafka and foreach sinks as at-least-once, and the foreachBatch sink as "Depends on the implementation". For a Delta table sink, the transaction log guarantees exactly-once processing "even when there are other streams or batch queries running concurrently against the table".

What happens if a write inside foreachBatch fails?

An error that escapes your function terminates the query, and query.exception() returns it. Restart the query with the same checkpoint location and it recovers its progress and continues where it left off. The interrupted batch runs again with the same batch ID, which is why every write in the function has to be idempotent.

How does Databricks approach writing to multiple sinks?

Databricks recommends multiple Structured Streaming writers, one per sink, "for best parallelization and throughput". Writing to several sinks inside foreachBatch serializes the writes, which can increase latency for each micro-batch. Each query must have a different checkpoint location; multiple queries should never share one. If you do write several Delta tables from foreachBatch, use txnAppId and txnVersion, and don't catch and retry exceptions locally. Databricks recommends letting the query fail fast and having an orchestrator such as Lakeflow Jobs retry the batch. Your function should also handle an empty micro-batch, as the isEmpty() check above does.
In Lakeflow Pipelines, sinks attach to an append_flow. A ForEachBatch sink, declared with @dp.foreach_batch_sink, can write each micro-batch to multiple destinations. Include df.persist or df.cache in the function so Databricks attempts to read the data only once.

FAQs

Can two streaming queries share a checkpoint location?

No. Each query needs a different checkpoint location, because the checkpoint records that query's offsets and commits.

Does foreachBatch work in continuous processing mode?

No. It relies on micro-batch execution. Use foreach in continuous mode.

Why is my source read twice inside foreachBatch?

Each write can recompute the micro-batch output, including the input read. Persist the batch DataFrame before the writes and unpersist it after them.