Spark Streaming checkpoint损坏致任务失败，求无重复处理恢复方案

针对你遇到的Spark Streaming任务因Checkpoint状态丢失、batchId重置导致的启动失败问题，以下是无需全量重跑数据的恢复步骤：

一、先备份Checkpoint目录 #

操作前务必完整备份现有Checkpoint目录，避免操作失误无法回滚：

dbutils.fs.cp("dbfs:/mnt/path/my_table/", "dbfs:/mnt/path/my_table_backup_202X05XX", recurse=True)

二、修正Checkpoint元数据（核心步骤） #

Spark Streaming的Checkpoint根目录下的metadata文件是状态记录的核心，里面包含latestBatchId（最后成功处理的批次ID）等关键信息：

1. 查看当前metadata内容：

dbutils.fs.head("dbfs:/mnt/path/my_table/metadata")

2. 编辑该文件，将latestBatchId修改为你找到的最后有效批次ID（即sources/0目录下的711），同时将currentBatchId设置为712（下一个待处理的批次ID）。确保JSON格式完全正确，无语法错误。
3. 将修改后的metadata文件覆盖回原Checkpoint目录。

三、同步Sources、Offsets、Commits目录状态 #

1. Sources目录：确认sources/0目录下存在711文件（已存在），无需额外操作；若任务启动后仍报错找不到对应批次文件，可复制711文件并重命名为报错提示的缺失文件（比如0），但优先以修正metadata为准。
2. Offsets目录：检查offsets/0目录下是否存在0到711的批次文件。若缺失最后几个批次的offset文件，可从备份目录中复制对应文件；若备份也没有，可手动构造offset文件（内容为该批次已处理的S3文件路径列表，格式与现有offset文件一致）。
3. Commits目录：确保commits/0目录下存在711文件（标记该批次已提交），若缺失则从备份复制或手动创建空文件（文件名需与批次ID一致）。

四、启动任务的参数调整 #

启动任务时添加以下参数，避免临时Checkpoint相关报错：

--conf spark.sql.streaming.forceDeleteTempCheckpointLocation=true

同时确认任务代码中的Schema与当前Parquet表的Schema一致（包含3月新增的列），避免Schema不匹配问题。

五、测试与验证 #

先启动单测试任务，观察：

• 任务是否正常启动，无找不到文件的报错
• 是否从批次712开始处理新数据
• 数据写入Parquet表是否正常
  测试通过后，再恢复生产环境的4个任务。

内容的提问来源于stack exchange，提问作者Martín Riccardi