Where is Spark structured streaming state of mapGroupsWithState stored?

user_2201566 0
Viewed 446

I know that the state is persisted at the checkpoint location as the state store. but I don't know while it's still in memory, where it's stored?

I created a Streaming job that uses mapGroupsWithState, but I see that storage memory used by executors is 0.

Does this mean that the state is stored in the execution memory? I can't know the amount of memory consumed by the state. not sure how to know if I need to increase the executor memory or not!

Also, is it possible to avoid checkpointing of the state at all and keep the state always in memory?

spark-structured-streaming apache-spark
2 Answers

As mapGroupsWithState is an aggregation it will be stored where all aggregations are kept within the lifetime of a Spark application: In the Execution Memory (as you have already assumed).

Looking at the signature of the method

def mapGroupsWithState[S: Encoder, U: Encoder](
      func: (K, Iterator[V], GroupState[S]) => U): Dataset[U]

you will notice that S is the type of the user-defined state. And this is where the state is managed.

As this will be sent to the executors it must be encodable to Spark SQL types. Therefore, you would typically use a case class in Scala or a Bean in Java. The GroupState is a typed wrapper object that provides methods to access and manage the state value.

It is crucial that you, as developer, also take care of how data gets removed from this state. Otherwise, your state will inevitably cause an OOM as it will only grow and never shrink.

If you do not enable checkpointing in your structured stream then nothing is stored but you have the drawback to loose your state during a failure. In case you have enabled checkpointing, e.g. to keep track of the input source, Spark will also store the current state into the checkpoint location.

user_122089100
user_29785440
Related