Failure Recovery Procedures

Datanest Digital — https://datanest.dev

Overview

Streaming pipelines fail. The difference between a minor incident and a major outage is how quickly and correctly you recover. This guide provides step-by-step recovery procedures for the most common failure scenarios in Databricks Structured Streaming and Delta Live Tables.

Each section follows the same structure: symptoms, root cause, impact assessment, recovery steps, and prevention measures.

Scenario 1: Checkpoint Corruption

Symptoms

• Stream fails on restart with StreamingQueryException
• Error messages referencing OffsetOutOfRange, FileNotFoundException in the checkpoint path, or metadata file is corrupted
• Stream starts but immediately stops after batch 0

Root Cause

• Hard cluster termination during a micro-batch commit
• Storage system returning partial writes (rare but possible with object storage under load)
• Manual modification or deletion of checkpoint files

Impact Assessment

Recovery Steps

Step 1: Stop the stream and back up the checkpoint.

for q in spark.streams.active:
    if q.name == "your_stream_name":
        q.stop()
        q.awaitTermination(timeout=60)

from datetime import datetime
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
dbutils.fs.cp(checkpoint_path, f"{checkpoint_path}_backup_{ts}", recurse=True)

Step 2: Diagnose the corruption.

def diagnose_checkpoint(path: str) -> dict:
    result = {}
    try:
        result["metadata"] = dbutils.fs.head(f"{path}/metadata", 1024)
    except:
        result["metadata"] = "MISSING_OR_CORRUPT"

    for subdir in ["offsets", "commits"]:
        try:
            files = dbutils.fs.ls(f"{path}/{subdir}")
            ids = sorted([int(f.name) for f in files])
            result[subdir] = {"count": len(ids), "range": f"{min(ids)}-{max(ids)}" if ids else "empty"}
        except:
            result[subdir] = "MISSING"

    return result

import json
print(json.dumps(diagnose_checkpoint(checkpoint_path), indent=2))

Step 3: Apply the appropriate repair.

• Missing commit file: Create an empty file at the missing batch ID (see config/checkpoint_management.md for details).
• Corrupt metadata: Regenerate the metadata JSON with the original stream UUID.
• Corrupt state: Delete the state/ directory; the stream will rebuild state from scratch.
• Irrecoverable: Archive the checkpoint and restart from startingOffsets.

Step 4: Restart the stream and verify.

# After repair, restart the stream
query = (
    stream_df.writeStream
    .option("checkpointLocation", checkpoint_path)
    .start()
)

# Verify the first few batches complete successfully
import time
time.sleep(30)
print(json.dumps(query.lastProgress, indent=2, default=str))

Prevention

• Store checkpoints on durable cloud object storage (ADLS Gen2, S3, GCS) — never ephemeral DBFS
• Enable spark.sql.streaming.minBatchesToRetain=100 for automatic checkpoint compaction
• Monitor checkpoint health using the dashboard queries in monitoring/streaming_dashboard.sql

Scenario 2: Schema Evolution

Symptoms

• AnalysisException: Cannot resolve column name after source schema changes
• mergeSchema errors on Delta writes
• Deserialization returns null for new fields; data silently lost

Root Cause

• Upstream producer adds, removes, or renames fields without coordination
• Schema registry version mismatch between producer and consumer
• Source table schema changed via DDL while stream is running

Impact Assessment

Recovery Steps

Step 1: Identify the schema change.

# Compare current source schema to the schema in the streaming query
source_df = spark.read.table("analytics.bronze.raw_events")
source_df.printSchema()

# Check the target table schema
target_df = spark.read.table("analytics.silver.processed_events")
target_df.printSchema()

Step 2: Handle the change based on type.

For new fields (additive change):

# Ensure mergeSchema is enabled on the sink
query = (
    stream_df.writeStream
    .option("mergeSchema", "true")
    .start()
)

For renamed or removed fields:

# Update the stream's transformation logic to map old name to new name
stream_df = stream_df.withColumnRenamed("old_field", "new_field")

# Or handle missing fields gracefully
from pyspark.sql.functions import lit
if "removed_field" not in stream_df.columns:
    stream_df = stream_df.withColumn("removed_field", lit(None).cast("string"))

For type changes:

# Cast the field to the expected type with error handling
from pyspark.sql.functions import col, when
stream_df = stream_df.withColumn(
    "amount",
    when(col("amount").cast("double").isNotNull(), col("amount").cast("double"))
    .otherwise(None)
)

Step 3: Reset the checkpoint if the schema change is incompatible with the existing state.

Stateful operations (aggregations, dedup) store data in the old schema. If the schema change affects columns used in state, you must reset:

# Archive old checkpoint
dbutils.fs.mv(checkpoint_path, f"{checkpoint_path}_pre_schema_change", recurse=True)

# Restart stream — will reprocess from startingOffsets

Step 4: Backfill if needed.

If the stream was down during the schema transition, use availableNow to catch up:

query = (
    stream_df.writeStream
    .trigger(availableNow=True)
    .option("checkpointLocation", checkpoint_path)
    .start()
)
query.awaitTermination()

Prevention

• Use a schema registry (Confluent, Azure Schema Registry) with compatibility enforcement
• Set mergeSchema=true on all streaming sinks as a default
• Test schema changes in staging before applying to production sources
• Add schema validation expectations in your DLT pipelines (see dlt/expectations_library.py)

Scenario 3: Out of Memory (OOM)

Symptoms

• Executor lost: ExecutorLostFailure (executor X exited caused by one of the running tasks) Reason: Container killed by YARN for exceeding memory limits
• java.lang.OutOfMemoryError: Java heap space or GC overhead limit exceeded
• Streaming query stalls with increasing batch durations before eventual failure

Root Cause

• State store growth exceeding executor memory (most common)
• Single micro-batch too large (spike in source data volume)
• Skewed partitions concentrating data on a single executor
• Insufficient memory overhead for RocksDB native memory

Impact Assessment

OOM failures cause the micro-batch to fail and retry. With checkpointing, no data is lost, but the pipeline stalls until the OOM condition is resolved. Repeated OOM on the driver can crash the entire cluster.

Recovery Steps

Step 1: Identify the memory consumer.

# Check state store size
progress = query.lastProgress
for op in progress.get("stateOperators", []):
    print(f"State rows: {op.get('numRowsTotal', 0)}")
    print(f"Memory used: {op.get('memoryUsedBytes', 0) / 1024 / 1024:.1f} MB")

# Check micro-batch size
print(f"Input rows: {progress.get('numInputRows', 0)}")
print(f"Batch size bytes: {progress.get('sources', [{}])[0].get('numInputRows', 0)}")

Step 2: Apply immediate mitigation.

For state store OOM — tighten the watermark:

# Reduce watermark from 24 hours to 1 hour (evicts old state faster)
stream_df = stream_df.withWatermark("event_timestamp", "1 hour")

For batch size OOM — limit data per trigger:

# Kafka: reduce max offsets per trigger
.option("maxOffsetsPerTrigger", "10000")  # was 100000

# Auto Loader: reduce max files per trigger
.option("cloudFiles.maxFilesPerTrigger", "100")  # was 1000

For executor OOM — increase memory:

spark.executor.memory=32g
spark.executor.memoryOverhead=8g
spark.sql.streaming.stateStore.rocksdb.blockCacheSizeMB=512

Step 3: Switch to RocksDB state store if using the default provider.

spark.sql.streaming.stateStore.providerClass=com.databricks.sql.streaming.state.RocksDBStateStoreProvider

RocksDB stores state on local SSD instead of the JVM heap, dramatically reducing OOM risk for stateful operations.

Step 4: Restart the stream.

After applying config changes, restart the cluster and the streaming query. The query will resume from the last committed checkpoint.

Prevention

• Always use RocksDB state store for stateful streaming operations
• Set maxOffsetsPerTrigger or equivalent to cap micro-batch size
• Monitor state store growth and set alerts (see monitoring/streaming_alerts.sql)
• Use memory-optimized instance types for stateful workloads (see config/autoscaling_config.md)
• Configure watermarks as tight as your late-arrival tolerance allows

Scenario 4: Data Source Unavailability

Symptoms

• KafkaException: Failed to construct kafka consumer or TimeoutException
• EventHubException: The connection was inactive for more than the allowed idle timeout
• Auto Loader stream stops progressing; no new files detected
• query.status shows isTriggerActive: false with no errors (silent stall)

Root Cause

• Source system outage (Kafka broker down, Event Hub throttled, storage account unreachable)
• Network partition between Databricks cluster and source
• Authentication credential expiration (SASL token, SAS token, service principal secret)
• Source topic/container deleted or renamed

Impact Assessment

Structured Streaming will retry indefinitely on transient source failures. The stream pauses but does not lose its checkpoint position. Data produced during the outage accumulates at the source and will be processed when connectivity resumes (assuming source retention is sufficient).

Recovery Steps

Step 1: Verify source availability.

# Kafka — test broker connectivity
import socket
broker_host, broker_port = kafka_brokers.split(",")[0].split(":")
try:
    sock = socket.create_connection((broker_host, int(broker_port)), timeout=10)
    sock.close()
    print("Kafka broker reachable")
except Exception as e:
    print(f"Kafka broker unreachable: {e}")

# Auto Loader — verify storage path
try:
    files = dbutils.fs.ls(source_path)
    print(f"Storage accessible: {len(files)} files found")
except Exception as e:
    print(f"Storage unreachable: {e}")

Step 2: Check for credential expiration.

# Test secret access
try:
    secret = dbutils.secrets.get(scope="kafka", key="sasl_password")
    print("Secret accessible (value redacted)")
except Exception as e:
    print(f"Secret access failed: {e}")

If credentials have expired, rotate them in the Databricks secret scope and restart the stream.

Step 3: Check the streaming query status.

for q in spark.streams.active:
    print(f"Query: {q.name}")
    print(f"  Active: {q.isActive}")
    print(f"  Status: {q.status}")
    if q.exception():
        print(f"  Exception: {q.exception()}")

Step 4: Handle extended outages.

If the source was down longer than its retention period, data may be lost. For Kafka:

# Check if offsets are still available
# If you get OffsetOutOfRange on restart, you must decide:
# Option A: Reset to earliest available offset (may cause duplicates)
.option("startingOffsets", "earliest")
# Note: this only applies to a NEW checkpoint. For existing checkpoints,
# set failOnDataLoss=false to skip past unavailable offsets:
.option("failOnDataLoss", "false")

Step 5: Restart and monitor catch-up.

After source recovery, monitor the stream's catch-up progress:

import time
for _ in range(10):
    time.sleep(30)
    progress = query.lastProgress
    if progress:
        input_rate = progress.get("inputRowsPerSecond", 0)
        process_rate = progress.get("processedRowsPerSecond", 0)
        print(f"Input: {input_rate:.0f} rows/sec | Processed: {process_rate:.0f} rows/sec")

When processedRowsPerSecond stabilizes near inputRowsPerSecond, the backlog is cleared.

Prevention

• Set failOnDataLoss=false for Kafka sources in production to survive offset expiration gracefully
• Configure source retention to exceed your maximum expected outage window (e.g., 7 days for Kafka, 7 days for Event Hubs)
• Use multiple Kafka brokers and Event Hub partitions for high availability
• Monitor source connectivity from the streaming dashboard
• Set up credential rotation alerts before expiration

Testing Recovery Procedures

Do not wait for a production failure to discover your recovery process is broken. Test these procedures regularly in a staging environment.

Test Matrix

Running a Recovery Drill

# 1. Record current stream position
pre_test_progress = query.lastProgress
pre_test_batch = pre_test_progress.get("batchId", 0)
print(f"Pre-test batch: {pre_test_batch}")

# 2. Stop the stream
query.stop()
query.awaitTermination(timeout=60)

# 3. Introduce the failure (e.g., delete a commit file)
dbutils.fs.rm(f"{checkpoint_path}/commits/{pre_test_batch}")

# 4. Restart the stream
query = stream_df.writeStream.option("checkpointLocation", checkpoint_path).start()

# 5. Verify recovery
import time
time.sleep(60)
post_test_progress = query.lastProgress
post_test_batch = post_test_progress.get("batchId", 0)
print(f"Post-test batch: {post_test_batch}")
assert post_test_batch >= pre_test_batch, "Stream did not recover"
print("Recovery test PASSED")

Recommended Schedule

• Monthly: Checkpoint corruption and full reset tests
• Quarterly: OOM simulation and schema evolution tests
• After infrastructure changes: Source connectivity and credential rotation tests
• Before go-live: Full test matrix execution

Quick Reference: Recovery Decision Tree

Stream failed?
├── Check query.exception()
│   ├── Checkpoint error? → See Scenario 1
│   ├── Schema/Analysis error? → See Scenario 2
│   ├── OOM / memory error? → See Scenario 3
│   └── Connection / timeout error? → See Scenario 4
│
├── Stream running but not progressing?
│   ├── query.status shows waiting for data → Source may be empty (normal)
│   ├── query.status shows processing → Batch is slow (see performance tuning guide)
│   └── No lastProgress updates → Stream is stuck; restart the query
│
└── Stream running but producing wrong results?
    ├── Duplicates appearing → Check dedup watermark and checkpoint health
    ├── Missing data → Check source retention and failOnDataLoss setting
    └── Wrong aggregations → State store may be corrupt; consider state reset

*Real-Time Streaming Toolkit — Datanest Digital — https://datanest.dev*