Spark Structured Streaming中如何获取正在处理的Delta表版本？

我希望监控Spark Structured Streaming正在处理的Delta表版本，尤其是当流通过startingVersion选项启动时。我理解设置该选项后，流会通过重放源表版本来增量读取源表的变更，以此了解当前运行流的状态。

我先准备了拥有3个版本（0、1、2）的源Delta表，再创建目标表，随后启动流执行合并操作，并通过StreamingQueryListener监听startOffset/endOffset中的reservoirVersion字段，该字段文档描述为“当前正在处理的表版本”，但实际日志显示的是3（源表最高版本为2），与预期不符。

请问如何正确获取流正在处理的源Delta表版本？另外，使用startingVersion时，流是否确实会从指定版本开始读取变更？在源表较大的场景下，我需要从特定版本重启流（使用不同检查点），因此需要明确该逻辑。

解决方案 #

1. 正确获取流处理的Delta表版本 #

你看到的reservoirVersion显示为3是正常行为——这个字段的实际含义是下一个待处理的版本，而非当前正在处理的版本。当源表最高版本为2时，流已经完成了0-2版本的处理，因此reservoirVersion会指向不存在的版本3，代表流等待后续新的版本提交。

要获取当前批次实际处理的版本范围，需要解析QueryProgress中源的startOffset和endOffset里的startVersion和endVersion字段：

• startVersion：当前批次开始处理的Delta表版本
• endVersion：当前批次结束处理的Delta表版本

示例代码（Scala）：

import org.apache.spark.sql.streaming.{StreamingQueryListener, QueryProgressEvent, QueryStartedEvent, QueryTerminatedEvent}

class DeltaVersionListener extends StreamingQueryListener {
  override def onQueryStarted(event: QueryStartedEvent): Unit = {}
  override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {}
  
  override def onQueryProgress(event: QueryProgressEvent): Unit = {
    event.progress.sources.headOption.foreach { source =>
      // 解析起始和结束版本
      val startVersion = source.startOffset.asInstanceOf[Map[String, String]]("startVersion").toLong
      val endVersion = source.endOffset.asInstanceOf[Map[String, String]]("endVersion").toLong
      println(s"当前批次处理的Delta版本范围：$startVersion ~ $endVersion")
    }
  }
}

// 注册监听器
spark.streams.addListener(new DeltaVersionListener)

2. startingVersion的生效逻辑确认 #

startingVersion参数确实会让流从指定版本开始读取增量变更：

• 当设置startingVersion=N时，流会跳过版本N之前的所有变更，仅处理版本N及之后的所有提交（包括版本N本身的修改）。
• 例如你源表有版本0、1、2，设置startingVersion=1，流会处理版本1和2的所有变更，不会处理版本0的数据。

重启流的注意事项 #

如果需要从特定版本重启流并使用新检查点，必须满足两个条件：

• 使用全新的检查点目录：流启动时会优先读取检查点中的偏移量记录，如果检查点已存在，startingVersion参数会被忽略。
• 明确指定startingVersion：在启动流时配置.option("startingVersion", "N")，确保流从目标版本开始消费。

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