IntermediateSpark · 14 min · 13 challenges
Structured Streaming: Intermediate
A streaming query is a process that runs for months, and nothing that runs for months runs uninterrupted. Deploys restart it. Spot instances vanish under it. A bad record kills it at 2am. The beginner tier gave you the loop: micro-batches over an unbounded table, paced by a trigger. This tier is about the loop dying mid-stride, because that is where streams earn or lose their keep. What did the engine remember about its position? What happens to the batch that was half written when the process died? Why did the restarted stream write yesterday's revenue rows twice, and whose fault is that? The answers live in one directory of small files, one contract with three signatures, and one distinction that splits every streaming query into two kinds. Every stream dies eventually; this tier is how yours come back clean.
Checkpointing
You can now explain what lives in a checkpoint and why a stream without one is lost on restart.
When a streaming query restarts, everything in memory is gone: the driver that was pacing the loop, the executors that were mid-task, the engine's knowledge of which offsets it had processed. If that knowledge lived only in memory, a restarted stream would face two bad options: start from the beginning of the source and reprocess everything, or start from the end and silently skip whatever arrived during the outage. The checkpoint is the third option: a durable record of the stream's position, written to reliable storage as the stream runs, so a restart resumes exactly where the crash interrupted.
The checkpointLocation you set on writeStream in the beginner tier points at a directory, and its contents are worth knowing by name because you will one day inspect them during an incident. The offsets directory holds one file per micro-batch, recording which source offsets that batch covers. The commits directory holds one file per completed batch. A metadata file identifies the query, and a state directory, which this lesson's last section introduces, holds the data stateful operations carry between batches. Small files, boring names, and together they are the entire difference between a stream and a hopeful loop.
| Checkpoint entry | What it records | When it is written |
|---|---|---|
| offsets/ | The exact source range batch N will cover | Before batch N runs |
| commits/ | Proof that batch N finished end to end | After batch N's output is written |
| metadata | The query's identity | Once, at first start |
| state/ | Carried data for stateful operations | Every batch, for stateful queries |
The ordering of those writes is the actual mechanism, and it is a write-ahead protocol. Before batch N processes anything, the engine durably writes the offsets file declaring what N will cover. Only after the batch's output reaches the sink does it write the commit file declaring N done. On restart, the engine reads both directories and compares: if the latest offsets file has a matching commit, the batch finished, resume from the next range. If the offsets file has no commit, the crash landed mid-batch, so the engine re-runs batch N over precisely the offset range the file recorded. Declared intent plus a replayable source makes the re-run deterministic: the same slice, reprocessed.
What the checkpoint is married to
A checkpoint is not a generic bookmark; it is specific to one query reading one source. Point a different query at the same checkpoint, or change the query's shape in incompatible ways, aggregating on a different key, say, and the engine refuses or misbehaves, because the recorded offsets and state describe a computation that no longer exists. And the innocent-looking rm on a checkpoint directory is a destructive act: the stream that starts afterward is a brand-new query with no history, which will either reprocess retained history or skip to the source's tail, depending on configuration. Teams learn this during the incident it causes.
TIP
Checkpoints belong on storage that survives every machine in the cluster: S3, HDFS, ADLS. A checkpoint on local disk protects you from a process crash and nothing else; the day the node itself dies, the stream's memory dies with it.
So the stream's memory is durable, its position declared before each step and confirmed after. But look again at the failure window that protocol leaves open: a crash after the sink write and before the commit file means the re-run will hand the sink rows it has already absorbed. Whether that becomes duplicate data is not the checkpoint's decision. It belongs to a contract between the source, the engine, and the sink, and that contract is next.
Exactly-Once as a Contract
You can now state the exactly-once contract: replayable source, checkpointed offsets, idempotent or transactional sink.
Delivery semantics come in three grades, and naming them precisely keeps every later argument short. At-most-once: an event affects the output zero or one times, so failures lose data. At-least-once: one or more times, so failures duplicate data. Exactly-once: the output ends up as if each event were processed precisely once, no losses, no doubles. Note the careful phrasing on the last one: it is a claim about the observable result, not about the machinery never retrying. Retries happen constantly in exactly-once systems; their effects just never land twice.
The claim that matters is that exactly-once is not a switch anywhere in Spark. It is a contract with three signatures, and Structured Streaming can only sign one of them. The source must be replayable: asked for an offset range a second time, it returns the same rows, which Kafka within retention and immutable files both do. The engine must track positions in a write-ahead checkpoint, which the previous section showed it does. And the sink must tolerate receiving the same batch twice without its contents landing twice, by being idempotent or transactional. All three, or the guarantee quietly degrades to at-least-once.
Replayable source
The same offset range, requested twice, yields the same rows. Kafka within retention and immutable files qualify; a forgetful firehose does not.
Checkpointed engine
Offsets written ahead, commits written after: Spark's signature, provided by the checkpoint protocol you already know.
Idempotent or transactional sink
A replayed batch must not double the output: the sink either recognizes the batch or commits it atomically exactly once.
Any signature missing
The pipeline is at-least-once at best, whatever the architecture slide says.
The sink's signature deserves the closest read, because it is the one most often forged. The file sink earns it with a transaction log: alongside the data files it maintains a _spark_metadata log recording, per batch ID, exactly which files that batch produced. A replayed batch writes its files again, but the log entry for that batch ID already exists, so readers of the directory never see the strays. Delta Lake earns it transactionally, committing the batch's rows and its ID atomically, and rejecting a second commit of the same ID. In both designs the batch ID, the number the engine assigns each micro-batch, is the linchpin: it is how a sink recognizes deja vu.
Now the sinks that do not sign. The Kafka sink is at-least-once: the engine may re-send a replayed batch's records to the topic, and Kafka appends them again; if downstream consumers cannot tolerate duplicates, deduplication becomes their job. And foreachBatch, the escape hatch from the beginner tier, hands you the batch DataFrame and its batch ID and makes idempotence entirely your problem. The honest implementations use the ID: overwrite a partition derived from it, or MERGE on keys so a replay updates rather than appends. Write a plain append inside foreachBatch and you have signed the contract in disappearing ink.
The daily-life version of this section is a checklist you can run against any pipeline in about a minute: can the source replay a range, is there a checkpoint on durable storage, and what exactly does the sink do when batch 4132 arrives a second time? Asking that third question by batch ID, concretely, is what separates engineers who have internalized the contract from those repeating the phrase exactly-once. And when the checklist fails, it fails in one characteristic way, visible in production as the most reported streaming bug there is. That bug gets the next section to itself.
The Dup After Restart
You can now diagnose duplicate rows after a stream restart and name the missing half of the contract.
The incident report is always the same shape. A stream ran fine for weeks, restarted for some ordinary reason, and now the downstream table has duplicates: the same order events twice, revenue double-counted for a 20-minute window, an analyst asking why Tuesday looks so good. Nobody changed the code. The restart is blamed, and the restart is innocent. The duplicates were latent in the pipeline's design from the day it shipped; the restart merely collected the debt.
Walk the timeline with the checkpoint protocol in hand. The engine writes the offsets file for batch N, declaring its slice. The batch runs and writes its output to the sink; the rows are now durably in the downstream system. And then, in the gap before the commit file for N lands, the process dies. On restart the engine does exactly what it should: offsets N exists, commits N does not, so batch N re-runs over the same slice and its output goes to the sink again. The engine cannot know whether the first write completed, so it must assume it did not. A sink that blindly appends now holds every row of batch N twice.
| Step | What happens | Where it lands |
|---|---|---|
| 1 | offsets/N written: batch N will cover this slice | Checkpoint |
| 2 | Batch N output written to the sink | Sink, durably |
| 3 | Crash, before commits/N is written | Nothing records step 2 happened |
| 4 | Restart: offsets N without commits N, replay batch N | Correct engine behavior |
| 5 | Same rows written to the sink again | Duplicates, unless the sink recognizes batch N |
Understand why the engine cannot fix this alone: the sink write and the commit-file write are two operations against two different systems, and Spark has no way to make them atomic from the outside. Some pair of them can always be split by a crash. The only clean resolution is the one the contract already named: the sink participates. Given a transactional sink like Delta, the batch's rows and its ID commit as one atomic action, and the replay's second commit of ID N is rejected. Given the file sink, the _spark_metadata log ignores the replay's files. Given foreachBatch, you write the recognition yourself, keyed on the batch ID: an overwrite or a MERGE, never a bare append.
The imposter with the same symptom
One other failure produces identical-looking duplicates with a completely different cause: an operational change that abandoned the checkpoint. A redeploy that pointed the query at a fresh checkpointLocation, or a well-meant cleanup that deleted the directory, creates a brand-new query with no memory, and if it starts from the earliest retained offsets it will reprocess days of history into a sink that already holds the results. Telling the two apart takes one look: replay-window duplicates span the minutes around a crash; checkpoint-loss duplicates span however far back retention reaches. The first is fixed in the sink; the second is fixed in the deploy process that treats checkpoints as disposable.
Multiple Choice
A stream restarted after a crash and the downstream table now holds each row from one micro-batch twice. Which part of the exactly-once contract was missing?
So the dup-after-restart is not a bug you patch but a design property you audit for: find every stream whose sink appends blindly, and fix the sink or accept documented at-least-once. Run the audit before the restart does it for you.
Output Modes
You can now pick the legal output mode for a given query and sink, and explain why the others are rejected.
The beginner tier named the output mode as a writeStream setting and moved on. Here is what it actually decides: of the result table's rows, which ones does the sink receive at the end of each micro-batch? For a query that only transforms, the question sounds trivial, but the moment a query aggregates, the result table contains rows that change as new data folds in, and delivering a changing row is a genuinely different act from delivering a finished one. The three modes are three answers, and each is only legal where its answer makes sense.
Append delivers only rows that are final: rows that will never change no matter what arrives later. For a pure filter-and-transform query, every output row is final the moment it is computed, so append is natural and cheap. For an aggregation it is a problem: the count for region EU changes every batch, so when is its row final? Never, unless something external promises that a group is closed. That something is the watermark, a declared bound on lateness that the advanced tier dissects; for now, the legality rule: append on an aggregation requires a watermark, and the engine emits each group's row once, when the watermark closes it.
Update delivers the rows that changed since the last batch. The EU count goes out every batch in which it moved, as a fresh version of the same logical row, which means the sink must be able to upsert: to overwrite its EU row rather than accumulate 500 of them. Complete delivers the entire result table, every batch, changed or not. That is only meaningful for aggregations, and only affordable when the result is small: a dashboard of counts for 200 countries, refreshed wholesale each batch, is a fine use. A complete-mode result keyed by user ID across millions of users re-delivers millions of rows per batch, forever, and is the mode chosen by mistake.
| Mode | What each batch delivers | Legal for | Sink it fits |
|---|---|---|---|
| append | Only rows that will never change again | Stateless queries; aggregations with a watermark | Plain appending sinks, files |
| update | Rows changed since the last batch | Aggregations without append's finality | Sinks that can upsert by key |
| complete | The whole result table, every batch | Aggregations only, small result cardinality | Sinks that can be wholly replaced |
The legality rules are enforced early, and that timing is a gift. Ask for complete mode on a non-aggregating query, or append on an unwatermarked aggregation, and the query fails at start() with an analysis error, before a single batch runs. Compare that with discovering the mismatch 3 weeks in. When the error message about modes and aggregations appears, it is not an obstacle to route around with a random different mode; it is the engine telling you your query's shape and your delivery intent disagree, and one of them is wrong.
In practice, choose the mode backward from the sink and forward from the query at once. Files in a lake can only append, so an aggregation feeding them needs a watermark and append mode, and consumers see each group once, when it closes. A key-value store or upsertable table pairs with update for live-refreshing aggregates. Complete is the special case reserved for small wholesale-refreshed results. And the fault line under all three, the reason aggregations keep complicating every rule in this section, is the distinction the whole lesson has been circling: whether the query must remember anything between batches. That split closes the tier.
Stateless vs Stateful
You can now split any streaming query into its stateless and stateful parts and predict the operational cost of each.
Every streaming query answers one question before any other: does producing correct output require remembering anything from previous batches? If no, the query is stateless. A filter, a select, a per-row transform, a parse: each row arrives, is processed on the spot, and is forgotten. Batch N needs nothing from batch N minus one. If yes, the query is stateful: a running count per region must carry the counts so far; a deduplication must remember which keys it has seen; a windowed aggregate holds partial sums for windows still open. That carried information is called state, and its existence changes the operational character of the stream more than any other property.
Where does state physically live? In a component called the state store: per-partition storage on the executors, keyed by the grouping key, persisted into the checkpoint's state directory every batch so a restart can rebuild it. The advanced tier opens the store's internals; what matters here is the structural fact that stateful operators write state to the checkpoint every single batch. A stateless query's checkpoint is a few kilobytes of offsets and commits. A stateful query's checkpoint grows with the number of distinct keys being tracked, and its per-batch work includes maintaining that store alongside processing the new slice.
•Stateless: nothing carried
- filter, select, map, parse: rows forgotten on sight
- Checkpoint holds offsets and commits, kilobytes
- Restarts are instant, nothing to rebuild
- Append mode, naturally: every row is final
•Stateful: memory with rent due
- Aggregations, dedup, windows, stream-stream joins
- State store written to the checkpoint every batch
- Restarts rebuild state before resuming
- Needs watermarks to keep state from growing forever
Classifying real queries takes practice, because the stateful ones do not always look it. dropDuplicates is the classic trap: it reads like a cleanup step, but to drop a duplicate you must remember every key ever seen, which without a bound means state that grows forever. A join between the stream and a static table is stateless with respect to the stream: each streaming row looks up the static side and moves on, nothing carried between batches. A join between two streams is deeply stateful, buffering rows from both sides awaiting matches, and it is enough of a subject that the advanced tier gives it a full section. The test is always the same: could this batch compute its output from its slice alone?
Why the split is the first question
Run the split on any proposed stream and you have costed it before writing code. The stateless version ships in an afternoon: any output mode that fits the sink, trivial checkpoint, restarts in seconds, scales by adding executors. The stateful version brings the full apparatus: state store maintenance every batch, restart time proportional to state size, memory pressure on executors, and an obligation to bound state growth with watermarks or the stream degrades over weeks. Neither is wrong; they are different prices. The recurring production mistake is pricing a stateful stream as if it were stateless because the code for both fits in 10 lines.
✓Do
- Classify every streaming query stateless or stateful before estimating its cost or its restart behavior.
- Give every stateful operator a bound: a watermark, a key space you can defend, or both.
- Keep the stateless portion of a pipeline stateless: push filters and parsing ahead of the first stateful operator.
- Check the checkpoint's state directory size in incident triage; unbounded growth there is a diagnosis.
✗Don't
- Don't add dropDuplicates to a stream casually; unbounded, it is a slow memory leak with correct output.
- Don't assume a stream-static join costs what a stream-stream join costs; only one of them carries state.
- Don't let an unbounded aggregation run in update mode just because the engine accepts it; growth arrives on schedule.
- Don't price a stateful stream by its line count; price it by its keys.
You now hold the full correctness story: a checkpoint that remembers position, a three-signature contract for exactly-once, the restart bug that finds the unsigned sinks, modes that define delivery, and the stateless-stateful split that prices everything. The advanced tier walks through the door this section opened: where state lives, how watermarks bound it, and what happens when the input outruns the stream.
❯❯❯PUTTING IT ALL TOGETHER
> Monday morning, finance flags that weekend revenue in the warehouse is inflated. The ingestion stream crashed on a spot reclaim Saturday night and auto-restarted cleanly, say the logs. The duplicated rows all carry timestamps from a 12-minute window around the crash.
The tight window around the crash tells you this is a replayed batch, not a checkpoint loss: retention-deep duplicates would span days, not minutes.
You read the checkpoint: the crash batch has an offsets file and no commit, so the engine replayed it on restart, exactly as designed.
The sink is a foreachBatch doing a bare append into the warehouse: the missing third signature of the exactly-once contract, latent since the day it shipped.
The fix is a MERGE keyed so a replayed batch updates instead of appends, and the postmortem action is an audit of every other stream whose sink appends blindly.
KEY TAKEAWAYS
The checkpoint is a write-ahead protocol: offsets declared before each batch, commits recorded after, on storage that outlives the cluster.
Exactly-once is a three-signature contract: replayable source, checkpointed engine, idempotent or transactional sink; any missing signature means at-least-once.
Duplicates after a restart mean the sink absorbed a replayed batch without recognizing its batch ID; the fix lives in the sink, not the engine.
Output modes decide which result rows each batch delivers: append needs finality, update needs an upsertable sink, complete needs a small result.
The stateless-stateful split sets a stream's real cost: state is memory with rent due, and it must be bounded or it grows forever.
Every stream dies eventually. The good ones come back without writing twice.
- Category
- Spark
- Difficulty
- intermediate
- Duration
- 14 minutes
- Challenges
- 13 hands-on challenges
Topics covered: Checkpointing, Exactly-Once as a Contract, The Dup After Restart, Output Modes, Stateless vs Stateful
Lesson Sections
- Checkpointing (concepts: paStreamProcessing)
When a streaming query restarts, everything in memory is gone: the driver that was pacing the loop, the executors that were mid-task, the engine's knowledge of which offsets it had processed. If that knowledge lived only in memory, a restarted stream would face two bad options: start from the beginning of the source and reprocess everything, or start from the end and silently skip whatever arrived during the outage. The checkpoint is the third option: a durable record of the stream's position, w
- Exactly-Once as a Contract (concepts: paStreamProcessing)
Delivery semantics come in three grades, and naming them precisely keeps every later argument short. At-most-once: an event affects the output zero or one times, so failures lose data. At-least-once: one or more times, so failures duplicate data. Exactly-once: the output ends up as if each event were processed precisely once, no losses, no doubles. Note the careful phrasing on the last one: it is a claim about the observable result, not about the machinery never retrying. Retries happen constant
- The Dup After Restart (concepts: paStreamProcessing)
The incident report is always the same shape. A stream ran fine for weeks, restarted for some ordinary reason, and now the downstream table has duplicates: the same order events twice, revenue double-counted for a 20-minute window, an analyst asking why Tuesday looks so good. Nobody changed the code. The restart is blamed, and the restart is innocent. The duplicates were latent in the pipeline's design from the day it shipped; the restart merely collected the debt. Walk the timeline with the che
- Output Modes (concepts: paStreamProcessing)
The beginner tier named the output mode as a writeStream setting and moved on. Here is what it actually decides: of the result table's rows, which ones does the sink receive at the end of each micro-batch? For a query that only transforms, the question sounds trivial, but the moment a query aggregates, the result table contains rows that change as new data folds in, and delivering a changing row is a genuinely different act from delivering a finished one. The three modes are three answers, and e
- Stateless vs Stateful (concepts: paMicroBatchVsTrue)
Every streaming query answers one question before any other: does producing correct output require remembering anything from previous batches? If no, the query is stateless. A filter, a select, a per-row transform, a parse: each row arrives, is processed on the spot, and is forgotten. Batch N needs nothing from batch N minus one. If yes, the query is stateful: a running count per region must carry the counts so far; a deduplication must remember which keys it has seen; a windowed aggregate holds