Apache Spark / Advanced: streaming and recoverable operation / Lesson 14
Checkpoints, sinks and delivery semantics
Connect replayable inputs, durable progress and sink idempotency before claiming an end-to-end guarantee.
On this page
Lesson goals
- Explain checkpoint identity and query recovery compatibility.
- Design a sink outcome protocol for repeated micro-batches.
A delivery guarantee belongs to the whole path
Structured Streaming coordinates incremental progress, but end-to-end behavior also depends on a replayable source and a sink that supports the required commit semantics. A testing socket source cannot provide the same recovery contract as a documented replayable source. A sink that appends every repeated request can duplicate results even when Spark restores its checkpoint correctly. Name the failure scenario: driver restart, sink response loss, source retention expiry and incompatible query change require different reasoning. Avoid calling every streaming application exactly-once merely because a checkpointLocation appears in its configuration.
Treat a checkpoint as owned query state
A checkpoint can store progress and state metadata needed to resume a query. Use an approved durable location dedicated to that query, with suitable access and lifecycle controls. The builder fragment shows a fictional location but starts no query. Sharing the same checkpoint between independent queries or deleting it as a routine fix can destroy recovery assumptions. Query changes have compatibility restrictions, particularly for stateful operator schemas and source or sink changes. A fresh checkpoint is a new recovery history and may replay data, so it requires deliberate source and sink reconciliation.
writer = (windowed.writeStream
.outputMode('update')
.option('checkpointLocation', 'lab-checkpoints/site-window-query-v1'))A writer builder only; no sink or start call is supplied. The checkpoint path is fictional and no storage is accessed.
foreachBatch needs an idempotent external protocol
foreachBatch exposes a micro-batch DataFrame and a batch identifier for custom handling. Its default write guarantee is at-least-once unless the application uses a suitable deduplication or transactional protocol. A batch identifier is useful only within a defined query identity and recovery history; resetting checkpoints can change that context. If a batch writes two unrelated destinations, completion of one does not atomically commit the other. Keep per-destination outcomes and reconcile partial success. Retrying an incremental UPDATE that adds totals can double-count, whereas a reviewed key-based replacement may be safely repeatable.
Outcome records distinguish unknown from rejected
The local ledger describes query q1 and batch17 with an unknown sink outcome after a lost response. The correct next action is reconciliation by the persisted operation identity, not automatically appending the batch again. A deduplication marker written before the payload can lose data after a failure; a marker written afterward can duplicate payload on replay. Use a supported atomic relation between marker and data or another proven idempotent sink design. The JSON is a teaching model, not a working implementation or a claim that Spark supplies this external ledger automatically.
{"query_identity":"q1","batch_id":17,"operation_key":"q1:17","sink_outcome":"unknown","next_step":"reconcile persisted operation before replay"}A valid local outcome model. It contains no external endpoint and performs no sink write.
Rehearse restart with the exact query and sink
An authorized lab should test interruption before sink commit, after commit but before acknowledgment, and between multiple destination writes. Verify source offsets, checkpoint recovery, state correctness and one intended business effect per operation identity. Also test source retention long enough for recovery and supported changes across versions. Continuous processing and asynchronous progress options have different restrictions and guarantees; do not transfer micro-batch claims to them casually. This course creates no checkpoint, launches no query and invokes no sink. Only the local ledger parser and outcome reasoning are validated, with no exactly-once engine test claimed.
Common errors and fixes
A lost sink response is treated as rollback
- Why it happens
- The client lost knowledge of a possibly committed operation.
- How to fix it
- Reconcile the persisted operation identity before replay.
- Prevent it next time
- Test commit-acknowledgment loss explicitly.
Deleting a checkpoint silently duplicates history
- Why it happens
- A new progress history was started without sink reconciliation.
- How to fix it
- Review source positions, query identity and replay policy before reset.
- Prevent it next time
- Protect checkpoint lifecycle and recovery ownership.
A prewritten deduplication marker loses payload
- Why it happens
- Marker and data were not committed atomically.
- How to fix it
- Use a supported transactional or proven idempotent sink protocol.
- Prevent it next time
- Rehearse every marker/payload failure window.
Check your understanding
1. What does sink_outcome="unknown" imply about replay?
Show worked solution
It means the application must discover the durable outcome before deciding whether replay is safe. A timeout is not proof that nothing was written. Use query and operation identity with the approved sink protocol, and retain the unknown state until reconciliation resolves it.
2. Why is writing a processed-batch marker before data insufficient?
Show worked solution
A failure after the marker but before data can cause replay to skip missing payload. Reversing the order can duplicate payload after another failure. The marker and effect need an atomic or otherwise proven idempotent relationship; ordering alone is not that guarantee.
Key takeaways
- Checkpointing and external sink atomicity are different guarantees.
- Unknown outcomes require reconciliation rather than blind replay.