Stateful Processing

Overview

Learn how to use stateful processing to track, update, and manage session states across streaming micro-batches.

What You Will Learn

In this lesson, you will learn:

Stateful Management: Tracking state across micro-batch boundaries.
mapGroupsWithState: Updating a state for a key.
State store backends: Configuring state repositories.

Detailed Concept Explanation

By default, Spark streaming queries operate record-by-record or batch-by-batch.

However, many real-world use cases (like session tracking or user clickstream analysis) require you to persist state for a key (like a user ID) across multiple micro-batches.

In Spark, you manage state using two main APIs:

mapGroupsWithState: Applies a custom function to update state for each group key, returning one output row per key.
flatMapGroupsWithState: Similar, but allows the function to output 0 or more rows per key.

Code Examples

Scala Implementation (Stateful APIs are highly optimized in Scala)

scala

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming._

val spark = SparkSession.builder().appName("StatefulScala").getOrCreate()
import spark.implicits._

case class UserEvent(userId: String, action: String)
case class UserSessionState(clickCount: Int)
case class UserSessionOutput(userId: String, totalClicks: Int)

val eventsStream = spark.readStream.format("rate").load()
  .selectExpr("CAST(value AS STRING) AS userId", "'click' AS action")
  .as[UserEvent]

// Define state update function
val updateState = (userId: String, events: Iterator[UserEvent], state: GroupState[UserSessionState]) => {
  val current = state.getOption.getOrElse(UserSessionState(0))
  val updatedCount = current.clickCount + events.size
  state.update(UserSessionState(updatedCount))
  UserSessionOutput(userId, updatedCount)
}

val sessions = eventsStream
  .groupByKey(_.userId)
  .mapGroupsWithState(GroupStateTimeout.NoTimeout())(updateState)

val query = sessions.writeStream.format("console").start()
query.awaitTermination(5000)
query.stop()

Expected Output

text

+------+-----------+
|userId|totalClicks|
+------+-----------+
|   101|          5|
+------+-----------+

Execution Plan Diagram (Python & Scala)

Execution Plan Diagram

SparkSession.builder

eventsStream.as[UserEvent]

groupByKey(_.userId)

mapGroupsWithState(updateState)

writeStream

start()

Common Mistakes

Infinite State Store Growth: Forgetting to configure state timeouts (GroupStateTimeout). Without timeouts, inactive keys will remain in the state store forever, leading to executor memory leaks and crashes.