Spark structured streaming — Checkpointing
What I’ve learned about spark streaming checkpointing, how it’s implemented, and how the implementation details of offsets might lead to surprising behavior.
(A story saved from the drafts)
Spark structured streaming is a great feature. As you might already know it can be utilized to stream from files, or in our specific case, from delta tables. But eventually, a failure occurs, and when it does, we need the stream to start from the same place we stopped. We certainly wouldn’t want to reread and reprocess years of data we’ve already processed. A stream needs fault tolerance, and for that, it needs to answer the question of identity (which stream is it) and location — where is it reading from on the stream (which is continuous by nature). So a way to reason about checkpoints might be: Checkpoints are the “state” of your stream.
We’ll initiate the actual stream when we trigger the writeStream and provide it (among other options) the `checkpointLocation`.
Note for this specific stream format (delta table) we can also provide the maxFilesPerTrigger option. This option controls, for file source, the “maximum number of new files to be considered in every trigger” (Kafka source has a similar maxOffsetsPerTrigger semantics, and file sources also have maxBytesPerTrigger which is a soft limit). Sometimes when you start a stream from a pre-existing file source with lots of existing files, you’ll want to limit the number of files to allow your cluster to process it, as I did. In our specific use case, I also needed to tune max shuffle partitions for this job.
Checkpoints implementation details and gotchas
The first gotcha that hit me was failing with the basic reasoning about checkpoints as stream state. I had multiple sources (think staging and production) streaming data into the same multitenant table, and naively I’ve used the same checkpointing folder, thinking it’s a different source so it won’t matter. I was wrong. Checkpoints are the “identity” and “state” of your stream, remember? This means that even if it’s a multi-tenant/multi-version Spark “table” is being streamed into. Each stream needs its own checkpointing folder, mixing them up was painful.
The second gotcha is less obvious and more tied to implementation details of the checkpoints abstractions, so let’s look at how the checkpoints are actually implemented and peak inside your checkpoints folder.
The checkpoints folder is HDFS Metada log that uses (distributed) storage for fault tolerance and reliability. If we look in it we can see this structure:
offsets/
commits/
Metadata
Inside the offsets and commits folders we’ll find ordered, numbered files (0 ->n), which serve as a WAL (write-ahead log) for the stream and are instances of OffsetSeq.
If it’s a working stream you might notice the offsets folder is almost always one number ahead of the commits folder, commits are committed at the end of the micro-batch, But (this is the important part) offsets are created at the start of the micro-batch as part of the construct next micro-batch phase of the run. This is an important implementation detail that has real implications and is my (not) favorite spark streaming gotcha.
Get Alon Nisser’s stories in your inbox
Join Medium for free to get updates from this writer.
Looking inside an offset file (offset 0 in this case)
v1
{“batchWatermarkMs”:0,”batchTimestampMs”:1636151706639,”conf”:{“spark.sql.streaming.stateStore.providerClass”:”org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider”,”spark.sql.streaming.join.stateFormatVersion”:”2",”spark.sql.streaming.stateStore.compression.codec”:”lz4",”spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion”:”2",”spark.sql.streaming.multipleWatermarkPolicy”:”min”,”spark.sql.streaming.aggregation.stateFormatVersion”:”2",”spark.sql.shuffle.partitions”:”200"}
{“sourceVersion”:1,”reservoirId”:”b8ae1fc7–6ea2–4f76–9b95–01c96056b808",”reservoirVersion”:0,”index”:24,”isStartingVersion”:true}
While this contains data concerning the streaming job, the source, etc we can see there are “spark userland” definitions saved here as part of the conf, for example: shuffle.partitions . We can also point to the index, resolving what should be processed in this offset, which was derived by the “maxFilesPerTrigger” option. To illustrate that we can look at the next offset file (1)
v2
{“batchWatermarkMs”:0,”batchTimestampMs”:1636151706639,”conf”:{“spark.sql.streaming.stateStore.providerClass”:”org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider”,”spark.sql.streaming.join.stateFormatVersion”:”2",”spark.sql.streaming.stateStore.compression.codec”:”lz4",”spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion”:”2",”spark.sql.streaming.multipleWatermarkPolicy”:”min”,”spark.sql.streaming.aggregation.stateFormatVersion”:”2",”spark.sql.shuffle.partitions”:”200"}}
{“sourceVersion”:1,”reservoirId”:”b8ae1fc7–6ea2–4f76–9b95–01c96056b808",”reservoirVersion”:0,”index”:49,”isStartingVersion”:true}
You can see the change in “index” from 24 to 49 => which correlates to our maxFilePerTrigger configuration as specified above: 25.
This is usually working as expected. Well unless you’re doing a tuneup of those parameters on your job. Because your tuneup would influence only the next written offset. If you’re trying to tune up a streaming job that chokes on the size of the data (as I did) preventing it from ever finishing the micro-batch. And you’re tuning the shuffle partitions and the max number of files options, they won’t help you unless you manually remove the offset containing this metadata!
It gets worse.. because some streams might do “stateful calculations” spark has a safe guard that shuffle.partitions can’t be changed after the stream starts and checkpoints are created. Spark would just ignore you trying to set this via spark.conf. So if you do need to change them you have two options:
- Start a new stream, and specify a starting point near the end of the current stream. For this to work you’ll your stream processing to be idempotent. or else you’ll have data corruption, as the streams “reconsumes” the same items.
- Manually edit (yes I’ve done this, it works, but be careful) the latest “offset” file and change the value
In both cases of course only if you’re not doing stateful calculations on your stream, just micro-batch processing.
I’d consider this a surprising behavior, I would expect the conf and index of an uncommitted offset to be recalculated if the configuration changed and the stream restarted. I have just spent almost a day playing with stream configurations of a streaming job choking on the size of data without any success until I realized nothing I do influences what’s happening and had to dig in to understand what’s happening and how to fix it.
Now, after understanding how offsets are managed we can also manually move the stream processing back if, for example, we introduced faulty logic and we don’t want to re-consume the stream from the start. We’ll need to decrement commits and offsets until we get to the correct position. While it might be better to use built-in options to control start (such as `startingVersion` or `startingTimestamp` with a delta table stream source or startingOffsets with Kafka source). sometimes it’s not supported by the source or the supported version isn’t suited for your needs.
Summing up
Checkpoints are a powerful and important tool with spark streaming, and you need to dig them. As happens quite often when working with spark, you’ll need to look at the implementation details to understand surprising behavior.
Further reading resources:
- Offsets and Metadata Checkpointing from The Internals of Spark Structured Streaming gitbook.
- Detailed discussion of checkpoints in spark structured streaming in the always insightful “waiting for code” blog.
- Spark structured streaming explanation about the configurations that can’t be changed with checkpoints for a stream
Special thanks to Yerachmiel Feltzman for his insights and corrections.
