How do I write a stream to a Delta table?

Summary

• Call writeStream on a streaming DataFrame with .format("delta"), a checkpointLocation option, and either .toTable("<table>") (Spark 3.1 and later) or .start("<path>"). Delta's transaction log gives exactly-once processing, even when other streams or batch jobs write to the same table.
• The Delta sink accepts two output modes: append, the default, which adds new rows, and complete, which replaces the table with the full result every trigger. It rejects update mode.
• Every streaming query needs its own checkpoint in a fault-tolerant, HDFS-compatible file system. If it lives inside the table directory, use a _-prefixed folder such as <table>/_checkpoints so VACUUM skips it.
• Triggers set the cadence: the default runs micro-batches back to back, processingTime runs them on a fixed interval, and availableNow=True processes everything available and then stops.
• For upserts, call a Delta MERGE inside foreachBatch. For plain writes inside foreachBatch, add the txnAppId and txnVersion options so a replayed batch is ignored.

How do I write a stream to a Delta table?

Use Structured Streaming's Delta sink. Point writeStream at the delta format, give the query a checkpoint location, and start it with toTable() for a catalog table or start() for a path. Start with append mode and toTable(). Reach for foreachBatch only when you need an upsert or another batch-only operation. In open-source Spark, configure Delta Lake on the session first: the delta-spark package, spark.sql.extensions set to io.delta.sql.DeltaSparkSessionExtension, and spark.sql.catalog.spark_catalog set to org.apache.spark.sql.delta.catalog.DeltaCatalog.

The basic append write

events = spark.readStream.table("raw_events")  # any streaming source works

query = (events.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/tmp/delta/events/_checkpoints/")
    .trigger(availableNow=True)
    .toTable("events"))

query.awaitTermination()

toTable("events") creates or appends to a catalog table. Swap in .start("/tmp/delta/events") to write to a path instead. With availableNow=True, the query processes everything that's available, possibly across several micro-batches, then stops, which suits a scheduled job. Drop the trigger to run micro-batches as soon as the previous one completes, or use .trigger(processingTime="1 minute") for a fixed interval.

Output modes the Delta sink supports

Mode	What reaches the table	Use it for
append (default)	Only rows added since the last trigger	Ingestion, events, logs
complete	The whole result table after every trigger, replacing the table	Small aggregates, such as event counts per customer
update	Not accepted; the sink raises "does not support Update output mode"	Use foreachBatch with MERGE instead

A complete-mode aggregate looks like events.groupBy("customerId").count() followed by the same writeStream chain with .outputMode("complete"). Because every trigger rewrites the table, keep complete mode for results that stay small.

Checkpoints: one per query, and keep them stable

The checkpoint records the range of offsets processed in each trigger plus any running aggregates, so a restarted query picks up where it stopped. That's also why two queries can't share one. Store it in a fault-tolerant, HDFS-compatible file system. If it sits inside the table's directory, name the folder with a leading underscore, as in /tmp/delta/events/_checkpoints/; VACUUM skips directories that begin with _.
Treat the query's shape as fixed once it has a checkpoint. Spark's recovery rules don't allow changing the number or type of input sources between restarts, or changing stateful operations such as streaming aggregations and deduplication.

Upserts with foreachBatch

foreachBatch calls your function once per micro-batch with two arguments, that batch's output as a static DataFrame and its unique ID, so you can run a Delta MERGE on it. The Delta docs pair this with update mode, which emits only the aggregates that changed since the last trigger:

from delta.tables import DeltaTable

def merge_counts(micro_df, batch_id):
    target = DeltaTable.forName(spark, "customer_counts")
    (target.alias("t")
        .merge(micro_df.alias("s"), "t.customerId = s.customerId")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute())

(events.groupBy("customerId").count()
    .writeStream
    .foreachBatch(merge_counts)
    .outputMode("update")
    .option("checkpointLocation", "/tmp/delta/customer_counts/_checkpoints/")
    .start())

By default foreachBatch gives at-least-once guarantees, and a failed batch can run again. For plain DataFrame writes inside the function, Delta Lake 2.0.0 and above offers two writer options: .option("txnAppId", app_id).option("txnVersion", batch_id). Delta uses the pair to spot a batch it has already committed and ignore the duplicate.

How Databricks approaches streaming writes to Delta

The same code runs on Databricks against Unity Catalog. Pass a three-level name such as main.streaming_examples.events_by_customer to toTable(). The checkpoint location can be a Unity Catalog volume path such as /Volumes/catalog/schema/volume/path. Auto Loader adds the cloudFiles source, which processes new files incrementally as they arrive in cloud storage. In Lakeflow Pipelines, a streaming table is a Delta table with added support for streaming or incremental processing, and the pipeline manages the checkpoint location for you. When one stream feeds several tables, Databricks recommends a separate streaming write per sink over a single foreachBatch. See Delta table streaming reads and writes.

FAQs

Can several streams write to the same Delta table?

Yes. Delta's transaction log keeps processing exactly-once even when other streams or batch queries write to the table concurrently. Each query still needs its own checkpoint.

How do I let new columns from the stream into the table?

Set .option("mergeSchema", "true") on the writeStream. Delta then adds columns that are present in the data but missing from the table.

What happens if I delete the checkpoint?

The query loses its record of processed offsets and starts over as a new query. If you use txnAppId, change it too, because batch IDs restart from 0 and Delta would otherwise skip the new batches as duplicates.