Sev1: Databricks streaming job falling behind Event Hubs — lag growing, data hours late

#Symptoms

• Event Hubs consumer lag growing; dashboards hours behind.
• inputRowsPerSecond > processedRowsPerSecond in stream progress.
• Batch duration much longer than the trigger interval.

Warning

Event Hubs retention is limited. If lag approaches retention, unread events will be deleted. Catching up is the priority; root-cause later.

#1. Read progress

import json
q = spark.streams.get("<query-id>")
p = q.lastProgress
print(json.dumps({k: p[k] for k in ["batchId", "numInputRows", "inputRowsPerSecond", "processedRowsPerSecond", "durationMs"]}, indent=2))

durationMs breaks batch time into addBatch (sink write), getBatch, latestOffset, queryPlanning, walCommit. The biggest part is where to look.

#2. Parallelism

With the Kafka endpoint, each Event Hubs partition maps to one Spark task by default. 8 partitions = at most 8 parallel read tasks, regardless of cluster size.

df = (spark.readStream.format("kafka")
  .option("kafka.bootstrap.servers", "evhns-prod.servicebus.windows.net:9093")
  .option("subscribe", "telemetry")
  .option("kafka.security.protocol", "SASL_SSL")
  .option("kafka.sasl.mechanism", "PLAIN")
  .option("kafka.sasl.jaas.config", jaas)
  .option("minPartitions", "64")             # split Kafka partitions into more Spark tasks
  .option("maxOffsetsPerTrigger", "2000000") # bounded batches for steady catch-up
  .load())

Long term, more Event Hubs partitions (Premium/Dedicated allow changing partition counts; Standard is fixed at creation) give more parallelism end to end.

#3. Slow sink (addBatch dominates)

• Delta writes with many small files: enable optimized writes / auto compaction on the target.
• foreachBatch MERGE into a large table: make sure the MERGE condition prunes (partition or clustering column in the ON clause), deduplicate the batch first.
• Writing to an external system (SQL DB, API) inside foreachBatch: batch the calls and scale the target.

ALTER TABLE main.bronze.telemetry SET TBLPROPERTIES ('delta.autoOptimize.optimizeWrite' = 'true', 'delta.autoOptimize.autoCompact' = 'true');

#4. State growth

Stateful operations (dedupe, aggregations, stream-stream joins) without watermarks grow state forever, slowing every batch. Add watermarks; use the RocksDB state store provider:

spark.conf.set("spark.sql.streaming.stateStore.providerClass",
               "com.databricks.sql.streaming.state.RocksDBStateStoreProvider")

Changing state configuration may require a new checkpoint — plan for it.

#5. Catch up

• Temporarily scale the cluster up (more cores than total partitions × 2 is a good start) and raise maxOffsetsPerTrigger.
• If downstream consumers only need recent data, a controlled restart from a later position (new checkpoint with startingOffsets by timestamp) plus a backfill from Event Hubs Capture files is an option — get agreement first.

#Prevent it next time

• Alert on processing rate < input rate for 10+ minutes and on consumer lag.
• Load test at peak event rate × 2.
• Event Hubs Capture on important hubs for replay beyond retention.