Spark Structured Streaming读取Kafka时偏移量重置及查询启动问题

看起来你遇到的核心问题是：Spark会复用隐式生成的checkpoint记录的偏移量，忽略你设置的startingOffsets参数，哪怕你没显式指定checkpoint目录。下面一步步帮你解决：

1. 先搞懂为什么startingOffsets没生效 #

Spark Structured Streaming的startingOffsets只有在全新查询第一次启动时才会生效。如果作业之前运行过，哪怕你没指定checkpointLocation，Spark在本地模式下会自动创建临时checkpoint目录（比如/tmp/temporary-xxxxxx这类随机命名的目录），重启时会从这个临时checkpoint里读取之前的偏移量，直接跳过你设置的初始偏移量。

2. 显式指定checkpoint目录并手动清理 #

解决这个问题的关键是自己掌控checkpoint的位置，这样每次重启前可以彻底删除它，强制Spark启动全新查询。

修改你的writeStream代码，添加checkpointLocation配置：

val query = kafkaStreamingDF 
  .writeStream 
  .format("console") 
  .option("checkpointLocation", "/tmp/spark-kafka-dev-checkpoint") // 选一个你能找到的本地目录
  .start() 

每次在IntelliJ重启作业前，打开终端执行删除命令：

rm -rf /tmp/spark-kafka-dev-checkpoint

这样重启后，Spark会找不到之前的checkpoint，就会遵守你设置的startingOffsets读取数据。

3. 可选：代码自动清理checkpoint（仅开发测试用） #

如果觉得手动删除麻烦，可以在作业启动时添加一段代码，自动清理指定的checkpoint目录（注意：生产环境绝对不要这么做，会丢失偏移量导致数据重复或丢失）：

import java.io.File
import org.apache.commons.io.FileUtils

// 定义checkpoint目录路径
val checkpointDir = new File("/tmp/spark-kafka-dev-checkpoint")
// 检查目录存在就删除
if (checkpointDir.exists()) {
  FileUtils.deleteDirectory(checkpointDir)
}

// 后续的writeStream指定这个目录
val query = kafkaStreamingDF 
  .writeStream 
  .format("console") 
  .option("checkpointLocation", checkpointDir.getAbsolutePath)
  .start() 

这样每次重启作业，都会自动清理旧的checkpoint，确保从初始偏移量开始读取。

4. 验证设置是否生效 #

重启作业后，查看日志里的类似条目：

INFO KafkaSource: Starting offset for partition [你的topic名称]-0 is Some(0)

如果看到这样的日志，说明已经成功从指定偏移量开始读取了。如果还是显示latest，请检查：

• startingOffsets的格式是否正确（比如指定特定分区时，topic名称要和subscribe的完全一致）
• checkpoint目录是否真的被彻底删除了
• 有没有拼写错误（比如startingOffsets写成了startingOffset）

补充：关于startingOffsets的正确用法 #

• 全局设置最早偏移量：.option("startingOffsets", "earliest")
• 指定特定topic的分区偏移量：

.option("startingOffsets", """{"your-topic-name":{"0":0,"1":0}}""")

注意这里的topic名称必须和你subscribe的完全匹配，否则这个配置不会生效，Spark会 fallback 到默认的latest偏移量。

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