Spark Structured Streaming Output Modes: Append vs Update vs Complete

The short version

No aggregation: use append. A running aggregate going into a sink that can upsert: use update. A small aggregate the consumer replaces wholesale: use complete. Windowed aggregates going into files: append plus a watermark, because the file sink accepts nothing else.

What an output mode controls

Structured Streaming treats a stream as an unbounded result table that grows with every micro-batch. The output mode, set with writeStream.outputMode(), answers one question: on each trigger, which rows of that result table does the sink receive?

Whether the query aggregates decides which modes are legal. An unsupported combination fails at start() with an AnalysisException naming the output mode, so the mistake surfaces before any data moves.

Append: new rows, each written once

outputMode("append") is the default. Only rows added to the result table since the last trigger are written, and each row is written exactly once and never revised. That suits queries whose rows never change: select, filter, withColumn, stream-static joins, and stream-stream joins.

An aggregation can use append only on an event-time window with a watermark. Spark emits a window's row once the watermark passes the window's end, when the count can no longer change. Without a watermark, Spark cannot know when a window is finished, so the query is rejected.

from pyspark.sql.functions import window

counts = (clicks
    .withWatermark("ts", "15 minutes")
    .groupBy(window("ts", "10 minutes"))
    .count())

counts.writeStream.outputMode("append").format("parquet") \
    .option("path", "/out/clicks").option("checkpointLocation", "/chk/clicks").start()

Update: only the rows that changed

outputMode("update") writes only the result rows inserted or changed in this trigger. If Germany's total moved and France's did not, only Germany is written. Aggregates appear immediately, with or without a watermark, and with a watermark old state is cleaned up. For a query with no aggregation, update behaves exactly like append.

The catch is the sink. It receives the same key again and again, so it must be able to overwrite or upsert. The file sink cannot, which leaves Kafka, foreachBatch (for example a MERGE into a table) or a debugging sink.

Complete: the whole result, every trigger

outputMode("complete") writes the entire result table on every trigger, changed or not, and the sink replaces what it had. It works for aggregation queries only, since keeping every unaggregated row would be infeasible. It is also the only mode in which a stream can be sorted, after an aggregation.

The cost: all aggregate state is kept forever. A watermark cannot drop old groups, because the full result has to be reproduced each time. Fine for a count per country, expensive for millions of growing keys.

Which query allows which mode

Sourced from the output modes and supported-operations sections of the Apache Spark Structured Streaming Programming Guide.

Sink support: why files mean append

• File sink (format("parquet"), or json, csv, orc, with a path): append only, exactly-once. Files cannot be rewritten row by row, so update and complete are refused.
• Kafka sink: all three modes, at-least-once.
• Console and memory sinks: debugging only. Output is collected on the driver and is not fault tolerant; the memory sink's table is named by queryName().
• foreachBatch: hands each micro-batch to your function as an ordinary DataFrame plus a unique batch ID, so any batch writer works, including several sinks or a MERGE.

def upsert_totals(batch_df, batch_id):
    batch_df.createOrReplaceTempView("updates")
    batch_df.sparkSession.sql("""
        MERGE INTO totals t USING updates u ON t.customer = u.customer
        WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *""")

(per_customer.writeStream
    .foreachBatch(upsert_totals)
    .outputMode("update")
    .option("checkpointLocation", "/chk/per_customer")
    .start())

Update mode suits this: each trigger merges only the customers that changed. foreachBatch is at-least-once by default; use the batch ID to make writes idempotent.

Decision signals for the exam

Structured Streaming is 10% of the Databricks Certified Associate Developer for Apache Spark exam in the official guide, and its writing objective names output modes and sinks directly. These signals settle most questions:

• Windowed aggregation into files: append with withWatermark.
• Running aggregates, sink can upsert: update.
• Small result the consumer replaces, or a sorted result: complete.
• Large or ever-growing set of keys: not complete. It rewrites, and keeps in state, every group forever.

FAQ