如何从快照自动过期的Iceberg表读取Spark Streaming流数据？

问题背景 #

使用Spark Streaming读取Iceberg表作为数据源，该表从Kafka接收数据，已配置压缩及旧快照自动过期的维护规则。初始读取代码如下：

spark.readStream
  .format("iceberg")
  .option("streaming-skip-delete-snapshots", "true")
  .load(s"${icebergConf.icebergTableQualifier}")

当旧快照过期后，流任务直接失败，报错信息：

Exception in thread "main" ERROR: [STREAM_FAILED] Query [id = c084caa6-0907-4781-b6a7-8f8991929f97, runId = 7781194d-298d-4929-be8e-c5407bd98566] terminated with exception: Cannot load current offset at snapshot 1615816462090596768, the snapshot was expired or removed

尝试过启用streaming-skip-delete-snapshots、通过stream-from-timestamp指定最新快照时间戳两种方案，均未解决问题。

问题根源 #

1. 快照依赖与偏移绑定：Spark Streaming读取Iceberg时，流任务的偏移会绑定到特定快照ID，一旦该快照被自动过期删除，任务重启或续跑时找不到对应快照就会报错。
2. 配置兼容性/代码错误：
   • streaming-skip-delete-snapshots仅在Iceberg 0.14及以上版本提供稳定支持，低版本中该选项可能不生效。
   • 原stream-from-timestamp代码存在错误：committed_at本身就是毫秒级时间戳（Long类型），无需通过SimpleDateFormat解析，错误的转换逻辑导致时间戳参数无效。

可行解决方案 #

1. 确保Iceberg版本兼容 #

升级Iceberg至0.14.0及以上版本，该版本完善了流读取时的快照过期处理逻辑，streaming-skip-delete-snapshots选项才能正常生效，让任务在发现依赖快照被删除时自动跳转到下一个可用快照继续处理。

2. 修正stream-from-timestamp代码逻辑 #

原代码中对committed_at的解析完全多余，直接提取Long型时间戳即可，同时新增streaming-auto-offset-reset选项，让任务在找不到偏移快照时自动从最新位置续跑：

if (spark.catalog.tableExists(s"${icebergConf.icebergTableQualifier}.snapshots")) {
  val latestSnapshotTimestampDF = spark.read
    .table(s"${icebergConf.icebergTableQualifier}.snapshots")
    .agg(F.max("committed_at").as("latest_snapshot_timestamp"))

  // 直接提取Long型时间戳，无需转换解析
  val latestSnapshotTimestamp = latestSnapshotTimestampDF
    .head()
    .getAs[Long]("latest_snapshot_timestamp")

  logger.info(s"Latest Snapshot Timestamp: ${latestSnapshotTimestamp}")

  spark.readStream
    .format("iceberg")
    .option("streaming-skip-delete-snapshots", "true")
    .option("stream-from-timestamp", latestSnapshotTimestamp.toString)
    .option("streaming-auto-offset-reset", "latest") // 偏移丢失时自动从最新快照开始
    .load(s"${icebergConf.icebergTableQualifier}")
}

3. 调整快照过期策略与流任务检查点配合 #

• 延长Iceberg快照过期时间，确保过期时间大于流任务可能的最大中断时长（比如流任务可能停机3天，就将快照过期时间设为7天），避免任务重启时依赖的快照已被清理。
• 启用Iceberg元数据清理的安全配置：

  # 提交后保留的元数据版本数
  write.metadata.previous-versions-max=10
  # 延迟删除已提交的元数据文件
  write.metadata.delete-after-commit.enabled=true
  write.metadata.delete-after-commit.delay=86400s
  

4. 启用流读取进度自动更新 #

添加streaming-read-snapshot-progress选项，让流任务定期将检查点中的快照信息更新为最新已处理的快照，减少对旧快照的依赖：

spark.readStream
  .format("iceberg")
  .option("streaming-skip-delete-snapshots", "true")
  .option("streaming-read-snapshot-progress", "true")
  .option("streaming-auto-offset-reset", "latest")
  .load(s"${icebergConf.icebergTableQualifier}")

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