Introducing Simplified State Tracking in Apache Spark™ Structured Streaming

Summary

• State Reader API: Apache Spark's State Reader API enables users to access and analyze internal state data from Structured Streaming applications, making streaming workloads easier to debug and troubleshoot.
• New capabilities: This video reviews new change feed and snapshot features that allow efficient analysis of state changes, enabling developers to understand application behavior without disrupting production workloads on the Databricks Data and AI platform.
• Production optimization: Learn practical techniques for using the State Reader API to debug complex streaming scenarios, optimize state management, and ensure reliability of real-time data pipelines at scale.

Introducing Simplified State Tracking in Apache Spark™ Structured Streaming

Talk by Craig Lukasik, Sr. SSA, Databricks
This presentation will review the new change feed and snapshot capabilities in Apache Spark™ Structured Streaming’s State Reader API. The State Reader API enables users to access and analyze Structured Streaming's internal state data. Readers will learn how to leverage the new features to debug, troubleshoot and analyze state changes efficiently, making streaming workloads easier to manage at scale.
Here’s more to explore:
Production ready data pipelines for analytics and AI: https://www.databricks.com/solutions/data-engineering
The Big Book of Data Engineering: https://www.databricks.com/resources/ebook/big-book-data-engineering-2nd-edition
See all the product announcements from Data + AI Summit: https://www.databricks.com/events/dataaisummit-2025-announcements

FAQs

What is the State Reader API and why is it important?

The State Reader API allows you to access and analyze the internal state data maintained by Apache Spark Structured Streaming. This enables debugging, troubleshooting, and understanding state changes without disrupting production workloads.

What are the new change feed and snapshot capabilities?

These features provide efficient ways to examine how state changes over time (change feed) and capture state at specific points (snapshot). They make it easier to develop and debug stateful streaming applications on the Databricks Data and AI platform.

What is Apache Spark?

Apache Spark is a distributed computing framework for large-scale data processing and analytics. It provides APIs in Python, Scala, SQL, and R for batch processing, streaming, machine learning, and graph computations—making it foundational to modern data platforms including Databricks.

What's the state of Apache Spark in 2026?

Apache Spark continues evolving with simplified APIs for complex use cases, improved performance through optimizations like Photon, and better integration with AI workloads. This video demonstrates state tracking improvements for structured streaming—showing Spark's ongoing maturation for production analytics and AI applications.

What's the future of Apache Spark?

Apache Spark's future includes simplified programming models, tighter AI integration, and continued performance optimization for scale. The simplified state tracking in structured streaming exemplifies this—reducing complexity for developers while enabling more sophisticated real-time analytics use cases.

Full transcript

All right. So, pretty cool conference so far. Um, I hope you're having a great time networking, learning stuff, getting super excited about data and AI. Um, all right. So, here's like the obligatory slide. I don't think I have any forwardlooking statements, but if I do, uh, take them with a grain of salt. subject to change, etc. , etc. Uh, for the sessions y'all have been attending, is everyone doing this for every session you attend these ratings?
Because I don't know if you've realized it, but like at data bricks, we're a bit obsessed about data. So, like, please feed us the data. Um, so I'm a specialist solution architect at Datab Bricks. Oh, hopefully nobody has epilepsy. And uh I help clients. I'm sort of a wingman to the essays and uh I help with special, you know, architecture planning or sometimes deep dive optimization and then I get to help do talks like this.
And today we're going to just talk about some recent improvements to the state reader API. And uh has anyone used the state reader API yet? No. No. Okay, cool. Maybe kind of 0. 5. So today's talk is on the uh like a new addition to it. We'll review like the state reader API uh basics. Um but uh before we do that, let's I want to use our imagination a little bit. Like we're indoors in this conference center and like I prefer to be outdoors personally.
So we're going to use our imagination. We're going to we're going to pretend that our work involves the outdoors. So, we're going to imagine that we work in the wildlife management sector. So, in the United States, there are about 1,200 species of freshwater fish and many of them can be found in rivers and our job is to help ensure the health and continuation of various species. Right? So, we're going to look at how Apache Spark stateful structure streaming can help.
And during the development of our stateful stream streaming application, we'll see how the state reader API's change feed can help with development, debugging, and troubleshooting. So, data bricks is a great place to run streaming workloads. um you know in the early days of Apache Spark you know that wasn't its main bread and butter but in the early days of data bricks you could say our bread and butter wasn't warehousing but you know we've come uh up to speed and surpassed many capabilities and I think when it comes to streaming workloads um you know this speaks for itself and with Apache Spark 4.
0 stateful streaming uh is greatly uh improved. So stateful streaming the new stateful operator is called transform with state. If you've used the previous stateful operator which was apply uh in pandas with state or flat mapaps group with state you could kind of do what you needed to do but it was painful. It was not super efficient. Um so our exploration will uh involve a little we'll touch on that a little bit but I recommend looking into the other sessions that uh cover transform with state or just search it uh we have some blogs and stuff on it.
All right so back to our mission. So our mission is to alert wildlife managers when interventions are needed to maintain optimal fish populations and health. So we want our system to dispatch wildlife and fisheries experts into the field to apply a variety of measures to help the fish. So egg distribution, fish stocking, etc. And uh so for instance, we want to generate an alert if there's an imbalance in the distribution of sex for a certain species, you know, which could impact future fish stocks.
So you need a good ratio of males and females to ensure future generations can thrive. And so this is a perfect match for a stateful streaming pipeline. So in our use case um this sort of madeup use case and made up you know genai image there. Imagine that we have hundreds of IoT sensors monitoring fish movements across various river segments. Now our sophisticated IoT sensors can detect species, size, age and sex of each fish.
So with potentially millions of data points per second arriving continuously across all river segments, we can use structured streaming to optimize the data processing and parallelize the computations across a Spark cluster. All right. So um raise your hand if you've been using Apache Sparks for less than two years. All right. All right. So, we got a couple couple folks here. So, that's good because we're going to want to just review some structured streaming concepts before we go on.
And it can get sort of really confusing. Um, for instance, how that Yeah, it looks fine. So, the first concept is partitions and this can be really confusing because the source system that you connect to may have partitions of its own. So if you know Kafka, Kafka is composed of one or more a Kafka topic is composed of one or more partitions. And if you know Kinesis, a Kinesis screen stream is composed of one or more shards.
So in the source system uh that partitioning is sort of one concept that's one partitioning scheme. But then there's Spark that has its own concept of the input partition and the mapping from the source to Spark isn't always one to one. So for instance, uh if you're using Kafka, there's an option called min partitions that allows you to have a single Spark input partition reading from multiple Kafka partitions. All right.
So, and in the rest of this talk, these neither of these partition concepts are what we're going to talk about. There's this other partition concept which is uh spark. sql. shuffle partitions. Does everybody know that configuration? Yeah. If not, it's a good one to like just become familiar with. Um so when data is organized based on keys for wide transformations such as aggregations or group buys or joins shuffle partitions come to play.
Uh so a stateful stream a stateful stream's number of partitions is also configured by that spark. sql. shuffle partitions. So that's something uh that's important to know especially if you're working at massive scale. and the default is 200. So you can think of that as the number of grouping buckets that helps the correct keys get distributed to the correct executor or worker node. And again, so for the rest of our talk, we'll be focusing on that concept of shuffle partitions.
Uh because in a stateful stream, that's a mechanism for parallelizing across the keys. And bear with me, I've got more comments that I need to There we go. scroll a little bit. All right. And then these next concepts I'll go through fairly quickly, but they're worth like uh learning, especially if you have something at scale um or if you're using that custom PIS spark uh data source that I mentioned earlier. These concepts are pretty important.
So offsets, this is a to-do list. So this is like the Spark offset. So it actually can still be confusing because Kafka has its own offsets. Um, Spark has its own offset concept when it comes to micro batches. So, when you're streaming in Spark, it's doing these little, you know, pulses of processing and each microb uh micro batch has an offset number. Um, if you if you've ever looked in the checkpoint directory. So, if you have a stream and you're persisting it somewhere, you have to specify a checkpoint directory and we're going to look at that a little bit. um that's where it you can see sort of the batches and then within that it may contain the offsets related to the source system.
So it can be confusing but like in in u most cases you don't have to necessarily worry about the contents of the checkpoint directory but for large um streams you know sometimes knowing these little details can be handy. Uh there's also another directory in the checkpoint directory called offsets or sorry called commit and this is sort of the done log or the uh you know just after a microbatch is processed and successfully written to the sync or destination uh the batch ID will be written to the commit log and this is helpful um like if a job fails and has to restart it's going to check this commit log to see if it has to basically reprocess the previous microbatch.
Uh and so it's essential for job level fault tolerance and exactly once processing. All right. So that's a little bit of background. We understand our mission. Um so for those partition concepts, you know, we could potentially partition our inbound data on river segment or maybe river segment and uh species. So with transform a state um uh and with the previous stateful operator your key can be a sort of a composite key and then since our IoT sensors include some onboard memory I don't know if you noticed it in the little picture there um uh that on onboard memory might have some historical data.
Uh the timestamp field is something that we could potentially use as the offset like the source offset. Um cool. So now these other concepts are sort of relevant to stateful streaming. Um so Apache Spark is microbatch processing engine operates on a small batch of records at a time and in our fish monitoring use case we need to maintain some state across those batches. So, uh, if we capture a sliding window of six hours of fish data, so remember we want to see if there's going to be sort of future populations.
Are enough females and males sort of swimming by? Uh, if you want to capture like a sliding window, you're going to need some stateful data captured. Um, so the state store serves as that set of like think of it as many databases. So it's backed by rock d rocks db and that's actually a configuration you have to set it's not the default but those are essentially many databases that are distributed across the uh workers and uh for each microbatch some data is persisted to cloud storage as well to ensure durability and fault tolerance of the stream.
Um and when we look at the directory we see there's a commits and offsets uh related to the microbatch level and the state directory is a little bit more complex. There we go. Where my laser pointer is not very bright. Um but you can see the directory structure is a little bit different than the commits or offsets. Right? So there's a subdirectory for each stateful operator and then subdirectories underneath that for each partition uh of data.
Yeah. So these are sort of the concepts that are useful. And now um now we're going to review the state state reader API sort of original features. So, this talk is eventually uh I'm gonna have to speed it up, but eventually we're going to get to the change feed. Um, but we're going to review like the initial uh state reader API. And the state reader API has sort of two uh formats. So, if you see the spark read. format, uh the first one we'll talk about is state metadata.
Um but draw your attention to the right hand side. So, that's sort of the code associated with uh what's stored in state. So we can see we have two chained operators here. So two stateful operators are chained chained together and um we can see them in the results here. So the nice thing about this state reader API is it's just another spark format and then once you load it so you load it from the checkpoint directory and then once you have that you have your data frame and uh again this is spa uh spark read format state- metadata and this is just the highle high level stuff.
So we get fields for um number of state partitions uh the batch ids and uh what else do we have there? Yeah the min batch and the max batch ID. So when it comes to the state store uh iterations um so you know the Apache Spark is running these microbatches all those microbatches have a batch ID there's only a hundred uh sort of state store uh history change logs maintained um let's see what else here 100 Yep and there's a hundred uh sort of historical versions the state kept because you don't want your state to grow unbounded forever.
Um, and there's also a snapshot uh, which we'll get to in a little bit that can be used. So, if you have to restore your stream, it will, um, process change logs, but if there's a snapshot, it'll use that and then process any change logs after that. So again we see here we have two stateful operators spread across eight partitions and uh the state store name uh or the operator name is is uh oh sorry the state store name in this case is just default.
So if you have a streamtost stream join uh you'll see an indication of which is it the left or the right side of the stream stream join but in our case it's just a straight sort of transformation. Um and since we're using um transform with state we see that here and then there's a little bit of extra data that we won't necessarily have to worry about here. Um and then the other uh state reader API format is just called state store.
So state or spark read. format state store. Uh the state store format helps us peak into the actual contents of the state store. So this is not the output of the stream. This is the content of those many databases that hold that intermediate data for your stateful stream. So this is like the like the the knowledge that's needed, you know, in between your inputs and output. So it's the contents of what's in Rox DB, if you will.
Um, and so looking at this, it looks it looks fine, right? So um the code looks good, the state store data looks good, but the uh this is hiding a bug. So there's a bug in my transformer state logic that if I were to just use this state reader API in development and debugging which is one of the main sort of purposes of it. Um I wouldn't spot the bug and we'll get to the bug in a minute or three minutes. Uh so today we're talking about the state reader API's change feed and that lets you look at the changes over time over the course of batches.
So here uh really the only difference uh is I've added this read change feed and I've told it what batch I wanted to start to read at. So you can see so sort of the cool thing is in my stateful operator in my transformer state logic I've got a composite key species and river segment. I've got a map. Um, and I thought I had another state. Oh, uh, that's right. So, there's an option. So, in transform with state, you can actually have multiple named state variables.
And so, when you look at the state reader API, you provide the um, oops, you provide the name of the state variable. So, the bug here, which is kind of hard to see. So this is supposed to be a change feed but when I look at this I see the same key has a number of updates to the state store for the same batch ID. So that means my stateful operator is inefficient. So ideally so even though the state store is fairly efficient to update um every update you do to the state operate to the to the state value will be noted in the change log.
So sort of the highlight of this talk is really that the change feed when you're debugging uh or troubleshooting can help you see errors like this. So that there is is what smells funny if you don't if you don't see it. So for the same key of my state variable, I really should be updating it once. Like that would be the most efficient like way to go about it. And this is the code. So this is the I'm not going to go through all this code, but as I look through this, I see, oh, I'm updating the state variables within a loop.
So every time my stateful operator has to process a microbatch of rows for a given key, I'm looping through and I'm updating state. So that's that's not great. So I um I don't have time to go through the live demo, but one thing I want to point out is this state store change feed. Don't confuse it with like the delta change feed or CDC or sedd. So it's not intended to be used like that. Um that state um is not something that you can stream off of.
So it's only a spark read. format state store. There's no support for the read stream. U but you could still use that state store API or the state reader API uh to run queries and those queries could be used to populate dashboards. Um there's one other feature that you probably will never need. It allows you to read the snapshot file. So I'm talking about these change logs here. There's also a periodic like full snapshot of what's in state that gets synced down to cloud storage.
You can use this um snapshot start batch ID, snapshot partition ID, and you can basically have it ignore subsequent snapshots and it'll just basically start at a snapshot and then use the change logs. Again, this feature you should never need. This is like really intended if you suspect there's something wrong with the the snapshot version of state and you want to go, you know, yell at us to that there's a bug. This would help give you evidence like, hey, I can bypass this snapshot and get a different outcome.
So, this was probably developed as they developed, you know, the state uh logic. Um, but for your purposes, probably never need it. Um, so again, we love data. We're hungry for feedback. Uh, help us make the future sessions better. And, um, so the change feed, and it's just an one option in the state reader API option read change feed true uh, helped us find a performance bug in our stateful operator during development.
And um so you can see how this could probably be used for production troubleshooting development uh etc. Like previously with a uh previous stateful operators and pre-spark 4. 0 you would have had a lot of like print logic in your stateful operators and it'd be pretty gnarly because you couldn't get to the state. So this state reader API lets you peek into those many databases that store that sort of intermediate data.
And uh that's it. Thank you for the time.

Learn more about the Databricks Data and AI platform.