Exactly-Once as a Contract

Concepts covered: Structured Streaming

Exactly-once describes what ends up in the sink: every input record affects the output once, even when a crash forces Spark to process a batch twice. Spark makes no promise to execute each batch a single time, and the checkpoint walk-through showed it re-running batch 41 by design. The guarantee is a contract between 3 parties, and it holds only while all 3 keep their side of it.

A source that can replay

A source is replayable if it can serve an old range on request. A message log that keeps records for 7 days can re-serve the exact offsets batch 41 read, provided the query comes back within those 7 days. The file source re-reads files for as long as nobody moves or deletes them. A socket source keeps nothing: a line read from the socket is gone, and no checkpoint can bring it back, which is why Spark's documentation limits it to testing.

Replay has an expiry date. If cleanup deletes trip files, or the log's retention window passes, before the query restarts, the range recorded in offsets/41 no longer exists and the contract fails at its first party.

A sink that can take a replay

A sink keeps its side in 1 of 2 ways: it is idempotent, or it is transactional. An idempotent write leaves the same result whether it runs once or twice. A plain append fails that test, since 634 rows appended twice is 1,268 rows. These 3 patterns pass it:

• Overwrite by batch: write batch 41 to a folder or partition named by its batch id, replacing whatever is already there.
• Upsert by key: match rows on trip_id, so a second copy of a trip rewrites the same row instead of adding one.
• Record the batch id: store the last batch id the target applied, and skip any batch at or below it.

Upserts and batch-id tracking need a target that supports them, such as a database table with a key. Overwriting by batch needs only a folder. foreachBatch calls your function with each batch's DataFrame and its batch id, and a re-run of batch 41 arrives with id 41 again. Overwriting by batch and recording the batch id both key on that repeated id. The upsert keys on trip_id instead, a business key that is the same in both copies of a trip. Here is write_payout_lines rewritten with the first pattern:

def write_payout_lines(batch_df, batch_id):
    (
        batch_df.write
        .mode("overwrite")
        .parquet(f"/lake/payout_lines/batch_id={batch_id}/")
    )

Transactional: Spark's file sink

A transactional sink makes a batch's rows and its batch id visible in a single atomic step: both or neither. Spark's file sink works this way. Tasks write part files into the output folder, then the driver writes _spark_metadata/41, a log entry listing exactly those files. A re-run of batch 41 finds that entry and skips the write. Spark jobs that read the folder take their file list from _spark_metadata, so part files left by a crashed attempt stay invisible to them; a tool that lists the folder directly still sees them.

A crash hits in the middle of a batch. Which pipeline still ends with each trip in the output once?

• File source, checkpoint, foreachBatch appending each batch over JDBC
• Socket source, checkpoint, Spark's file sink with _spark_metadata
• File source, no checkpointLocation, foreachBatch overwriting by batch id
• File source, checkpoint, foreachBatch overwriting a folder per batch id

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.