Output Modes

Concepts covered: Structured Streaming

Each micro-batch updates a result table: the answer the query would give if it ran as a batch job over every row received so far. The output mode decides which rows of that result table go to the sink after each batch.

Mode	Sends to the sink after each batch	Fits
append	Rows added to the result table this batch, each sent once and never revised	Queries whose result rows never change once they exist
update	Rows added or changed this batch	Aggregations feeding a sink that can overwrite by key
complete	The whole result table, every batch	Small aggregations, such as totals for a few dozen cities

One query, 3 modes

Take a running count of completed trips per driver. In batch 1, driver d7 completes 2 trips and d9 completes 1. Batch 2 adds 1 more for d7 and 2 for a new driver, d4, and batch 3 adds 1 for d9.

Batch	Result table after it	update sends	complete sends
1	d7=2, d9=1	d7=2, d9=1	d7=2, d9=1
2	d7=3, d9=1, d4=2	d7=3, d4=2	d7=3, d9=1, d4=2
3	d7=3, d9=2, d4=2	d9=2	d7=3, d9=2, d4=2

Append has no column because Spark refuses it for this query. After batch 1 it would have to send d7=2 as a final row, and batch 2 shows it wasn't final. A running count changes whenever its driver finishes another trip, so no row of it is ever safe to emit as permanent. Spark accepts append on an aggregation only when the query also states how late an event may arrive, which lets Spark decide when a group is closed; that mechanism has a lesson of its own later in the track.

Complete is legal and wasteful at this shape. In batch 3 it resends d7=3 and d4=2, which did not change, and the table it resends grows with every new driver, so output per batch climbs for as long as the query runs. Update sent 1 row for the same batch.

Which modes a query may use

Query shape	append	update	complete
No aggregation: filter, select, withColumn	Allowed	Allowed, same output as append	Rejected: the result table would be every row ever read
Aggregation with no time bound	Rejected: groups never close	Allowed	Allowed
orderBy with no aggregation	Rejected	Rejected	Rejected
Aggregation followed by orderBy	Rejected	Rejected	Allowed

Spark checks the mode against the plan when you call start(), before any data is read, and raises an AnalysisException. A sort needs the full result in hand, which only complete mode provides, so orderBy is legal there and nowhere else. The 2 rejections you will meet most often read:

Messages raised at start()

Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark
Complete output mode not supported when there are no streaming aggregations on streaming DataFrames/Datasets

The sink gets a vote

A mode the query allows must also be one the sink accepts. The file sink writes immutable files and can't revise a row it has already written, so it takes append only. foreachBatch takes all 3 modes and hands your function the rows the mode selects, leaving it to your code to apply them correctly. The message-log sink also takes all 3, publishing each changed row as a new record for its consumers to reconcile.

This query writes per-driver counts to the file sink and sets no output mode. Which outputMode lets start() succeed?

counts = (
    trips
    .filter(F.col("status") == "completed")
    .groupBy("driver_id")
    .count()
)

query = (
    counts.writeStream
    .format("parquet")
    .option("path", "/lake/driver_counts/")
    .option("checkpointLocation", "/chk/driver_counts/")
    .start()
)

append, the default, because the file sink accepts append
update, because only changed counts are written each batch
None: the sink needs append and the aggregation forbids it
complete, because the query contains a streaming aggregation

About this lesson

This section is part of the Structured Streaming: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.