Spark Structured Streaming损坏后如何从Delta表最新版本启动并适配检查点

方案1：修改checkpoint偏移量直接对齐Delta最新版本 #

不需要删除完整checkpoint或delta事务日志，仅调整偏移量记录即可：

• 先查询Delta表当前最新版本：执行DESCRIBE HISTORY <你的Delta表名> LIMIT 1，获取返回结果中的version值，记为latest_version
• 进入你作业配置的checkpoint存储路径，找到offsets子目录，打开目录中序号最大的文件（即作业最后一次成功提交的偏移量记录）
• 找到文件中Delta源对应的reservoirVersion字段，将其值修改为刚才查询到的latest_version后保存
• 直接重启作业即可，作业会读取修改后的偏移量从最新版本开始消费，原有checkpoint中存储的状态计算数据（如聚合、join状态）可完整保留

方案2：配置启动参数自动跳过失效版本 #

如果是Delta 2.0及以上版本，可以直接添加作业启动配置，不需要手动修改文件：

spark.conf.set("spark.databricks.delta.streaming.source.ignoreDeletedVersions", "true")
spark.conf.set("spark.sql.streaming.fileSource.log.deletion", "false")

该配置会自动忽略Delta源中已经不存在的快照版本，自动将消费起始点对齐到当前可用的最新版本，不会再报delta log与checkpoint不匹配、快照不存在的错误。

方案3：显式指定起始版本重建偏移记录 #

如果前两个方案无法生效，仅需要删除checkpoint下的两个子目录即可，不需要删除完整checkpoint：

• 保留checkpoint下的state目录（存储状态计算数据），仅删除offsets和commits两个子目录
• 读取Delta流时显式指定起始版本参数：

// Scala示例
val deltaStream = spark.readStream
  .format("delta")
  .option("startingVersion", "latest")
  .load("<Delta表路径>")

• 重启作业后会自动生成新的偏移记录，从最新版本开始消费，原有状态数据不会丢失

• 以上方案均基于允许少量数据丢失的前提，如果需要严格保障数据不丢，需要先回溯delta历史日志修复与checkpoint偏移的对应关系
• 如果你的作业没有状态计算逻辑（无聚合、join等有状态算子），也可以直接删除完整checkpoint目录重启，操作更简单

内容的提问来源于stack exchange，提问作者user7454972