ExamFinch.com My Study

Study guide

2.2 Custom & Stateful Streaming Operations

Practice this exam

Databricks Certified Data Engineer Professional Exam

Track your progress as you read

Finished sections get a check mark.

1 Chapter 1: Developing Code for Data Processing using Python and SQL - Core Spark SQL & PySpark

1.1 Complex PySpark Transformations & DataFrame API 1.2 Advanced Spark SQL Patterns 1.3 Streaming Workloads with Structured Streaming 1.4 Lakeflow & Delta Live Tables (DLT) Declarative Pipelines

2 Chapter 2: Developing Code for Data Processing using Python and SQL - Streaming & Advanced Patterns

2.1 Advanced Delta Lake Features 2.2 Custom & Stateful Streaming Operations 2.3 Python & PySpark Code Modularization 2.4 User-Defined Functions (UDFs) & Performance Impact

3 Chapter 3: Data Ingestion & Acquisition

3.1 Auto Loader (cloudFiles) Production Ingestion 3.2 Change Data Capture (CDC) Ingestion Patterns 3.3 Batch & Streaming Connectors

4 Chapter 4: Data Transformation, Cleansing, and Quality

4.1 Data Validation & Quality Frameworks 4.2 Data Cleansing, Deduplication & Standardization 4.3 Silver and Gold Medallion Layer Refinement 4.4 Handling Schema Drift and Evolution

5 Chapter 5: Data Sharing and Federation

5.1 Delta Sharing Architecture & Governance 5.2 Lakehouse Federation

6 Chapter 6: Monitoring and Alerting & Observability

6.1 System Tables & Observability 6.2 Databricks Workflows & Job Monitoring 6.3 Spark UI & Query Execution Diagnostics 6.4 Data Quality & Pipeline Alerting

7 Chapter 7: Cost & Performance Optimisation - Data Layout & Compute

7.1 Data Storage & Layout Optimization 7.2 Compute Choice & Cluster Sizing 7.3 Photon Engine & Query Acceleration

8 Chapter 8: Cost & Performance Optimisation - Skew & Cost Governance

8.1 Partitioning & Data Skew Remediation 8.2 Cost Governance & DBU Optimization

9 Chapter 9: Ensuring Data Security and Compliance

9.1 Unity Catalog Access Control & Least Privilege 9.2 Dynamic Data Masking & Row/Column-Level Security 9.3 Secrets Management & Network Security 9.4 Compliance, Audit & Data Privacy

10 Chapter 10: Data Governance in Unity Catalog

10.1 Unity Catalog Governance Architecture 10.2 Data Lineage & Tagging in Unity Catalog 10.3 Managed vs External Tables & Storage Governance

11 Chapter 11: Debugging and Deploying - Asset Bundles & CI/CD

11.1 Databricks Asset Bundles (DABs) & Infrastructure as Code 11.2 CI/CD Automation & Databricks CLI / REST API 11.3 Troubleshooting Pipeline & Job Failures 11.4 Code Versioning & Databricks Git Integration

12 Chapter 12: Data Modelling & Lakehouse Analytics

12.1 Medallion Architecture Design Patterns 12.2 Dimensional Modeling in the Lakehouse 12.3 Analytics & BI Optimization

2.2 Custom & Stateful Streaming Operations

Key Takeaways Explain This Topic AI

Stateful streaming requires maintaining state across micro-batches, governed by the spark.sql.streaming.stateStore.providerClass setting.

RocksDB state store is recommended for stateful operations with large states that exceed executor memory, as it spills to local disk.

Stream-stream joins require watermarks on both sides of the join and a time range constraint to bound the state size and prevent OutOfMemory errors.

The flatMapGroupsWithState API provides granular control over state evolution, timeouts, and handling of late data in custom Python streaming logic.

Timeouts in custom stateful processing allow the application to cleanly remove state and emit final outputs when an entity is inactive.

Last updated: July 2026

Structured Streaming in PySpark allows for near real-time data processing. While stateless transformations (like map, filter, or standard selects) process each micro-batch independently, many complex applications require stateful processing. Stateful streaming tracks data across micro-batches for operations like aggregations, deduplication, stream-stream joins, and complex event processing.

Stateful Stream Processing Mechanics

When a stream performs a stateful operation (e.g., counting events per user over a tumbling window), Spark must remember previous data. This "memory" is stored in a State Store . During each micro-batch, Spark reads the previous state from the store, updates it with new incoming data, and writes the updated state back to the store.

Because state can grow indefinitely, stateful streams are highly susceptible to Out-Of-Memory (OOM) errors. Spark uses watermarks to bound state size. A watermark defines how late data can be before it is ignored. Once the watermark passes a certain time window, Spark drops the state for that window, freeing up resources. The watermark is calculated as the maximum event time seen so far minus the specified watermark delay.

State Store Configuration

By default, Spark uses an HDFS-backed state store (technically, in-memory state that is checkpointed to distributed storage). The default provider keeps state entirely in the executor's JVM memory. This is fast but risky for workloads with massive state (e.g., tracking millions of unique user sessions over a 30-day window). Memory constraints often lead to garbage collection pauses and application crashes.

For large-scale production workloads, the RocksDB state store is highly recommended. RocksDB maintains state locally on the executor's disk (spilling out of memory) while continuing to checkpoint to distributed storage for fault tolerance. RocksDB is an embeddable persistent key-value store optimized for fast storage environments like SSDs.

  • To enable RocksDB, configure the Spark session: spark.conf.set(
  • "spark.sql.streaming.stateStore.providerClass",
  • "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider"
  • )

This single configuration change can stabilize streaming applications that frequently crash due to memory pressure from large state. RocksDB handles state management much more effectively by utilizing disk space when memory limits are approached. It inherently supports versioning and snapshots, allowing Spark to reliably recover state during micro-batch failures.

Stream-Stream Joins

Joining two streaming DataFrames is one of the most complex operations in Structured Streaming. Because data on both sides is continuously arriving, Spark must buffer records from both streams in state until matching records arrive. If unconstrained, the state for both streams would grow indefinitely.

To prevent the state from growing infinitely, a stream-stream join must have three constraints:

Watermark on Stream A : Defines how late records from Stream A can arrive.

Watermark on Stream B : Defines how late records from Stream B can arrive.

Time Range Constraint in the Join Condition : Specifies the temporal relationship between the two streams (e.g., an ad click must happen within 1 hour of the ad impression).

  • Example: impressions_with_watermark = impressions_df \
  • .withWatermark("impressionTime", "2 hours")
  • clicks_with_watermark = clicks_df \
  • .withWatermark("clickTime", "3 hours")
  • joined_df = impressions_with_watermark.join(
  • clicks_with_watermark,
  • expr("""
  • clickAdId = impressionAdId AND
  • clickTime >= impressionTime AND
  • clickTime <= impressionTime + interval 1 hour
  • """)
  • )

In this example, Spark knows exactly how long to hold an impression in state. If an impression arrives at 12:00 PM, a matching click must arrive between 12:00 PM and 1:00 PM. Considering the watermarks, Spark can safely discard the impression state after a definitive boundary, preventing unbounded state growth. The exact state cleanup delay is dynamically computed using the watermarks from both streams and the time range constraint.

Custom Stateful Processing: flatMapGroupsWithState

While SQL aggregations and joins are declarative, some use cases require custom business logic that doesn't fit standard SQL functions. For instance, you might need to track a complex user journey, implement custom sessionization, or emit alerts when a specific sequence of events occurs.

For these scenarios, PySpark provides applyInPandasWithState (the Python equivalent of the Scala mapGroupsWithState / flatMapGroupsWithState APIs).

This API allows you to define a Python function that takes:

  • A grouping key (e.g., user_id)
  • An iterator of new rows for that key

A state object that can be read, updated, or cleared def process_user_events(key, pdf_iter, state):

  • # Read previous state
  • current_state = state.get()

if current_state is None:

current_state = {"login_count": 0, "last_event": None}

# Process new events in pandas dataframe iterator

for pdf in pdf_iter:

current_state["login_count"] += len(pdf)

  • # Update state
  • state.update(current_state)
  • # Yield output
  • import pandas as pd
  • yield pd.DataFrame({"user_id": [key[0]], "total_logins": [current_state["login_count"]]})

Crucially, this API allows you to configure Timeouts . You can set a processing time timeout or event time timeout. If no new data arrives for a key within the timeout period, your function is invoked one last time with an empty iterator, giving you a chance to emit a final record (e.g., "session closed") and clear the state using state.remove() . Properly clearing state is mandatory to avoid OOM errors in custom stateful streaming. The flexibility of this API allows data engineers to build sophisticated state machines directly within the streaming pipeline, managing complex workflows in real time.

This level of granular control is essential when building enterprise-grade real-time systems where standard aggregations fall short and custom session management, deduplication, and anomaly detection are required. This level of granular control is essential when building enterprise-grade real-time systems where standard aggregations fall short and custom session management, deduplication, and anomaly detection are required. This level of granular control is essential when building enterprise-grade real-time systems where standard aggregations fall short and custom session management, deduplication, and anomaly detection are required.