Structured Streaming

## The result-table model Structured Streaming asks you to think of the stream as a table that keeps growing. Each trigger appends the newly arrived records to a conceptual *input table*, Spark re-evaluates your query against the whole input table, and the answer is a conceptual *result table*. Spark does not literally materialise either one — the plan is executed incrementally, and only the state needed to keep the answer correct is retained. The **output mode** is the contract for which slice of that result table gets handed to the sink each trigger. That framing explains why the three modes exist. For a stateless query the result table only ever grows, so "what is new" is unambiguous. For an aggregation, an existing row can be *revised* when a later record lands in its group, and the sink now needs to be told whether it will receive revisions, only final values, or a full refresh. ## Append `outputMode("append")` is the default. It emits only rows added since the previous trigger **that are guaranteed never to change again**. For `select`, `filter`, `withColumn`, `explode` and other stateless transformations every output row qualifies the moment it is produced, so append behaves the way an ordinary batch write does. For a streaming aggregation the guarantee is the hard part. `groupBy(window($"ts", "10 minutes")).count()` cannot finalise a window's count until Spark believes no further records for that window will arrive — which is exactly what `withWatermark` declares. Write a streaming aggregation in append mode with no watermark and Spark refuses to start the query, reporting that append output mode is not supported for streaming aggregations without a watermark. With a watermark, the window's row is buffered in state and emitted once — after the watermark passes the window's end. The consequence is latency: a 10-minute window with a 5-minute watermark delay produces its first output roughly 15 minutes of event time after the window opened. Stream–stream outer joins have the same shape: the NULL-padded row can only be emitted once the watermark proves no match will arrive, so those joins require a watermark and a time-range condition. ## Update `outputMode("update")` emits every result row whose value changed during this trigger, and nothing else. Rows that were untouched are not re-sent. For an aggregation this means the same key is written repeatedly with a growing count — a partial answer, then a better one, then the final one. That is the lowest-latency mode for aggregations: you see a window's running count seconds after it opens rather than waiting for the watermark. The cost lands on the sink. Because the same key arrives many times, the sink must be an upsert (a key-value store, a `MERGE` in a table sink, a `foreachBatch` that writes by primary key) or the downstream reader must be tolerant of superseded values. Appending update-mode output to a plain file sink would give you every intermediate count as a separate row. A watermark is optional in update mode, but omitting it means aggregation state is never evicted and grows for the life of the query. ## Complete `outputMode("complete")` rewrites the entire result table on every trigger. It is only legal when the query contains an aggregation — there is no meaningful "whole table" for a projection over an unbounded stream. Because Spark must be able to re-emit every group forever, it retains all aggregation state regardless of any watermark you set; a watermark does not evict state in complete mode. That makes complete mode a small-cardinality tool: totals by country, counts by status code, a dashboard-shaped summary. Use it for per-user or per-session keys and state grows without bound and every trigger rewrites an ever-larger output. ## Sink compatibility The mode you want is not always available. The file sink supports append only, because files already committed cannot be rewritten in place. The Kafka sink accepts all three, though complete mode republishes the full table into the topic each trigger. The `console` and `memory` sinks accept all three and are debugging tools. `foreach`/`foreachBatch` accept all three and push the semantics onto your code — `foreachBatch` hands you a normal batch DataFrame plus the `batchId`, which is the usual way to implement an upsert for update mode. ## Practical guidance Pick append when the consumer wants immutable, final facts and can tolerate watermark latency; update when you want low-latency revisions and the sink can upsert; complete only for a small, dashboard-sized result. Treat the output mode as part of the query definition rather than a runtime knob: for a stateful query it interacts with the checkpointed state layout, so validate before flipping it against an existing checkpoint rather than assuming a restart will absorb the change.