Structured Streaming I — the micro-batch engine and exactly-once

Spark commits offsets durably after the write. A crash before the commit redoes the batch from the same offsets. A crash after the commit starts the next batch cleanly. This order is what makes recovery exactly-once with an idempotent/transactional sink.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#recovering-from-failures-with-checkpointing

The checkpoint stores committed offsets. A fresh directory looks like a query that never ran, so the query ingests everything again. This is the most common cause of duplicates on restart. Reuse the same checkpoint.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#recovering-from-failures-with-checkpointing

append emits each new result row once. It is the natural mode for a map-like (non-aggregating) stream. complete is valid only for aggregations (it rewrites the whole result in each batch).

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#output-modes

Without a watermark, any group may still receive more input. So its result is never final and Spark cannot append it. The modes complete and update work. A watermark (SPK-11) lets windows finalize, and then append becomes valid.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#output-modes

availableNow processes everything that is currently available and then stops. It still honors the checkpoint. It is good for tests (no waiting for wall-clock time) and for incremental batch jobs.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#triggers

Exactly-once needs the sink to be idempotent or transactional. A plain append on a redo writes twice. The file and Iceberg table sinks track committed batches to avoid this.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#recovering-from-failures-with-checkpointing

The watermark of the incremental model and the committed offsets of the stream both answer 'what is new since last time'. Streaming manages the offsets for you in the checkpoint. It does not rely on a column predicate that you maintain.

Source: https://spark.apache.org/docs/4.0.0/structured-streaming-programming-guide.html#basic-concepts