10.2 Stateless UDFs, Stateful Operators, and StateStores

Databricks Certified Associate Developer for Apache Spark — Free Study Guide 2026 Exam

Track your progress as you read

Finished sections get a check mark.

1 Chapter 1: Spark Architecture and Execution

1.1 Spark Benefits, Modules, and DataFrame/Dataset Concepts 1.2 Driver, Executors, and Cluster Managers 1.3 Execution Hierarchy: Applications, Jobs, Stages, and Tasks 1.4 Lazy Evaluation, Transformations vs Actions, and Resilient Lineage

2 Chapter 2: Shuffles, Memory and Persistence

2.1 Narrow vs Wide Dependencies and the Shuffle Mechanism 2.2 Memory Management, Fault Tolerance, and Garbage Collection 2.3 Caching, Persistence, and StorageLevels

3 Chapter 3: Spark SQL Sessions, Tables and Views

3.1 SparkSession, Catalogs, Databases, and Namespaces 3.2 Managed vs External Tables and CTAS 3.3 Views: Temporary vs Global Temporary Views

4 Chapter 4: SQL Sources and Stored Layout

4.1 JDBC Reads, Writes, and Parallelism 4.2 Direct File SQL, Save Modes, and Persistent Layout

5 Chapter 5: SQL Queries and Analytics

5.1 SQL Queries, CTEs, Subqueries, and Built-in Expressions 5.2 Built-in SQL Expressions and an Integrated Query 5.3 Window Functions and Complex Joins in Spark SQL 5.4 SQL Join Semantics and Integrated Analytics

6 Chapter 6: DataFrame Construction and Columns

6.1 DataFrame Construction, Schemas, and StructTypes 6.2 Column Operations: Selecting, Renaming, Deriving, and Casting 6.3 Complex Nested Types and User-Defined Functions

7 Chapter 7: DataFrame Validation and Inspection

7.1 Row Operations: Filtering, Sorting, and Deduplication 7.2 Handling Missing and Null Data 7.3 Schema Inspection, Row Iteration, and List Conversion

8 Chapter 8: Aggregations and Dates

8.1 Grouping, Multi-Dimensional Aggregations, and Pivoting 8.2 Dates, Unix Epoch, and Time-Zone Conversions

9 Chapter 9: DataFrame Joins and File IO

9.1 DataFrame Joins and Combining Datasets 9.2 Data Sources, SaveModes, and Partitioning

10 Chapter 10: Shared Variables and Stateful Processing

10.1 Broadcast Variables, Accumulators, and Closures 10.2 Stateless UDFs, Stateful Operators, and StateStores

11 Chapter 11: Troubleshooting and Tuning

11.1 Diagnosing Performance with the Spark UI & Execution Plans 11.2 Data Skew and Partition Tuning: repartition vs coalesce 11.3 Join Optimization and Broadcast Hash Joins

12 Chapter 12: Structured Streaming

12.1 Streaming Architecture, Sources, and Sinks 12.2 Output Modes, Triggers, and Checkpointing 12.3 Event-Time Processing, Watermarking, and Windowing 12.4 Streaming Deduplication With and Without Watermarks

13 Chapter 13: Spark Connect and Deployment

13.1 Spark Connect Architecture & Client-Server Decoupling 13.2 Working with Spark Connect: Connections and Best Practices

14 Chapter 14: Pandas API on Spark

14.1 Pandas API on Apache Spark: Core Architecture and Syntax 14.2 Pandas-on-Spark Operations, Indexing, and Optimization

10.2 Stateless UDFs, Stateful Operators, and StateStores

Key Takeaways Explain This Topic AI

A Python global or object field in a UDF is not a reliable distributed StateStore.

A streaming aggregation, deduplication or stream-stream join can retain state across batches.

StateStores version task state and use durable checkpoint storage for recovery; cache is a different mechanism.

PySpark applyInPandasWithState requires an explicit state/output schema and supported output/timeout configuration.

Last updated: October 2026

Decide whether the requirement needs history

A stateless UDF can validate a transaction from that row's fields alone: normalize a code, classify an amount, or check a custom format. Its output does not depend on which transaction happened in a previous batch. A stateful operator instead maintains information across input groups or batches, such as running customer totals, seen event identifiers, window counts or unmatched join records.

The distinction is semantic, not whether the function is written as a Python class. A module-level dictionary inside a scalar UDF is not reliably shared, partitioned or checkpointed. Tasks may run on different executors, Python workers can restart, and failures can replay input. Ordinary UDF instances provide no durable customer-history contract.

• Stateless validation example from pyspark.sql import functions as F
• from pyspark.sql.types import BooleanType

@F.udf(returnType=BooleanType())

def valid_reference(value):

if value is None:

• return False
• return value.startswith("R-") and len(value) == 8

checked = transactions.withColumn("valid_reference", valid_reference("reference"))

The function handles nulls itself and returns a declared BooleanType. It can run on batch or supported streaming DataFrame projections without accumulating earlier references. Prefer built-in string/regex expressions if they express the same rule; the example illustrates a scalar UDF contract. Registering it with spark.udf.register makes it SQL-callable but does not add persistent state.

Built-in stateful operations

A streaming groupBy("customer_id").count() retains a count per key. Streaming dropDuplicates remembers identifiers seen in previous batches. A stream-stream join buffers records while waiting for possible counterparts. These operators use Spark-managed state. Their memory and disk cost depends on key cardinality, retained event history and cleanup policies.

A watermark can let an eligible event-time operator finalize old state, while an application-specific timeout can define expiry for custom state. They are different mechanisms. Processing-time timeout reasons about elapsed processing time; event-time timeout depends on watermark advancement, which may stop if relevant data does not advance. Neither is an ordinary Python sleep or a promise of an exact wall-clock callback deadline.

What a StateStore provides

A StateStore is Spark's versioned keyed storage for stateful streaming tasks. Each micro-batch reads and updates an appropriate state version. Durable checkpoint data enables recovery after failure and helps replay a failed batch consistently. State is partitioned with the operator; it is not one giant Python dictionary on the driver.

The default HDFS-backed provider uses executor memory with checkpoint files in the configured fault-tolerant filesystem. "HDFS-backed" names the provider and does not require the checkpoint path to be an actual HDFS cluster. The RocksDB provider uses native storage and checkpoint mechanisms, often reducing JVM heap pressure for large state. RocksDB still uses memory, local disk and durable checkpoint IO; it does not make state free or eliminate all OOM risks.

• Mechanism Purpose Recovery/lifetime
• DataFrame cache Reuse computed partitions Can be evicted and recomputed
• Broadcast variable Read-only lookup reuse Not customer-history state
• StateStore Versioned stateful streaming state Recovered through its query checkpoint
• External table Persisted output/business data Governed by its provider and write semantics
• Custom keyed state in PySpark

Spark 3.5 provides applyInPandasWithState on grouped streaming DataFrames. The function receives a key, an iterator of pandas batches and a GroupState object. It yields pandas output batches, and the user declares both outputStructType and stateStructType. Do not assume the Scala/Java mapGroupsWithState signature is a direct Python DataFrame method. import pandas as pd

from pyspark.sql.streaming.state import GroupStateTimeout

def running_count(key, batches, state):

• previous = state.get[0] if state.exists else 0
• current = previous + sum(len(pdf) for pdf in batches)
• state.update((current,))
• yield pd.DataFrame({"customer_id": [key[0]], "n": [current]})

• counts = events.groupBy("customer_id").applyInPandasWithState(
• running_count,
• outputStructType="customer_id STRING, n LONG",
• stateStructType="n LONG",
• outputMode="Update",
• timeoutConf=GroupStateTimeout.NoTimeout
• )

This illustrative example assumes a string customer_id, counts every input row, uses a one-field tuple state and retains it indefinitely. The output writer must support Update. NoTimeout is intentional here: a production high-cardinality stream needs a retention policy. A timeout-aware function must distinguish a timeout invocation from new data and deliberately call state.remove or set new timeout metadata.

Recovery, compatibility and output

State schema and grouping keys are part of checkpoint compatibility. Changing a key type or adding an incompatible state field is not the same as adding a simple projection. Keep separate durable checkpoint locations for separate queries, and plan migrations/replay when changing stateful logic. Deleting a checkpoint loses the history of processed offsets and state.

State recovery does not guarantee an external callback is exactly-once. A retried output can duplicate a database insert unless the sink handles it. Keep state transitions and output semantics deliberate, monitor state row counts and memory/progress metrics, and test null keys, multiple input batches, key expiry and recovery with the supported runtime. See Structured Streaming state and arbitrary stateful operations .