Spark Structured Streaming从Kafka重置消费偏移量问题咨询

我之前也踩过一模一样的坑！你只删除HDFS上的Checkpoint目录还不够，因为Spark Structured Streaming和Kafka交互时，偏移量会被存在两个地方：

1. HDFS的Checkpoint目录：Spark用来保存流处理的状态和偏移量
2. Kafka内部的__consumer_offsets主题：Kafka自身会记录消费者组的历史偏移量

当你重启应用时，如果Kafka里对应的消费者组还有历史偏移量，哪怕Checkpoint删了，Spark还是会优先用Kafka里保存的偏移量，直接跳过startingOffset = earliest的设置。

给你几个靠谱的解决方案：

方案1：更换消费者组ID（最省心） #

在你的Kafka读取配置里手动指定一个group.id，每次需要从头消费时，换一个新的组名就行。比如：

val df = spark 
  .readStream 
  .format("kafka") 
  .option("kafka.bootstrap.servers", fromKafkaServers) 
  .option("subscribe", topicName) 
  .option("startingOffset", "earliest") 
  .option("group.id", "my-consumer-group-20240520") // 每次从头消费就修改这个值
  .load()

新的消费者组在Kafka里没有历史偏移量，重启后就会严格按照startingOffset的设置从头拉取数据。

方案2：清除Kafka里的消费者组偏移量 #

如果你不想换组名，可以用Kafka的命令行工具手动删除对应组的偏移量：

# 1. 先列出所有消费者组，找到你的应用对应的组
kafka-consumer-groups.sh --bootstrap-server <你的Kafka集群地址> --list

# 2. 查看该组的偏移量状态，确认是你要清理的目标
kafka-consumer-groups.sh --bootstrap-server <你的Kafka集群地址> --describe --group <你的消费者组ID>

# 3. 删除该组的所有偏移量
kafka-consumer-groups.sh --bootstrap-server <你的Kafka集群地址> --delete --group <你的消费者组ID>

执行完这步后，记得彻底删除HDFS上的Checkpoint目录，然后重启应用，就能从头消费了。

注意事项 #

• 清理Checkpoint时，一定要确保流应用已经完全停止，避免残留的状态文件干扰重启后的逻辑；
• 如果你的应用是集群模式运行，确认HDFS上的Checkpoint目录被彻底删除（HDFS是分布式存储，删一次就全集群生效）。

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