Sev1: Databricks streaming job won't restart after a code change — checkpoint incompatibility and safe recovery

#Symptoms

• After deploying new code, the stream fails at start with errors about incompatible state schema, a changed number of shuffle partitions, or a missing source in the checkpoint.
• Data in the target stops flowing; lag on Event Hubs/Kafka grows.

#1. Read the error

Typical messages (wording varies by runtime):

#2. Changes that are safe vs unsafe

Safe with the same checkpoint (generally):

• Changing filters, projections and map-like transformations on stateless queries.
• Changing the trigger interval.

Changing the output path of a file sink is not safe with the same checkpoint.

Unsafe (need a new checkpoint):

• Changing grouping keys, aggregation functions, window definitions, watermark-dependent state, dropDuplicates keys, stream-stream join conditions.
• Adding or removing input sources.

Check the Structured Streaming "recovery semantics after changes" documentation for your runtime.

#3. Fastest fix: roll back the code

If you can, redeploy the previous version and restart with the existing checkpoint. The stream resumes where it stopped with no data loss. Then plan the change properly.

#4. If the change must go in: new checkpoint, controlled start

You need to start a new checkpoint at the exact point the old one stopped, so nothing is skipped or duplicated.

Source is a Delta table: read the last processed version from the old checkpoint offsets and start from the next one:

# list offset files; the highest number is the last planned batch, check commits/ for the last finished one
display(dbutils.fs.ls("abfss://chk@stlake.dfs.core.windows.net/orders_agg/offsets/"))
display(dbutils.fs.ls("abfss://chk@stlake.dfs.core.windows.net/orders_agg/commits/"))
print(dbutils.fs.head("abfss://chk@stlake.dfs.core.windows.net/orders_agg/offsets/<n>"))

(spark.readStream.format("delta")
  .option("startingVersion", "<last_processed_version + 1>")
  .table("prod.raw.orders")
  ...)

Source is Kafka/Event Hubs: take the partition offsets from the old checkpoint's latest offsets/<n> file and pass them as startingOffsets (Kafka JSON format) or the Event Hubs starting positions.

Caution

Stateful aggregations restart with empty state. Windows that were open when you switched will be incomplete. Either backfill those windows from the source with a batch job, or start the new stream slightly earlier and make the sink idempotent (MERGE by key) so overlaps don't duplicate.

#5. Make the sink idempotent

With foreachBatch and a MERGE, reprocessing the same records is harmless:

def upsert(batch_df, batch_id):
    batch_df.createOrReplaceTempView("b")
    batch_df.sparkSession.sql("""
      MERGE INTO prod.sales.orders_agg t USING b s
      ON t.window_start = s.window_start AND t.store_id = s.store_id
      WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *""")

(df.writeStream.foreachBatch(upsert)
   .option("checkpointLocation", "abfss://chk@stlake.dfs.core.windows.net/orders_agg_v2/")
   .start())

#6. Keep the old checkpoint

Never delete the old checkpoint during the incident. Rename the new one with a version suffix (_v2). If the new approach fails, you can still roll back.

#Prevent it next time

• Version checkpoint paths in code (.../orders_agg_v3/) and change the version deliberately with a migration plan.
• Review streaming changes against the safe/unsafe list before merge.
• Use idempotent sinks for every production stream.
• Test restarts with the production checkpoint copy in staging.