Spark 3.3.0集成Kafka 3.2.0：Spark停止时如何终止数据流防丢失

你的问题核心在于启用了Kafka自动提交offset，这会导致Spark还没完成数据处理，Kafka就提前提交了offset。一旦Spark意外停止，重启后会从已提交的offset开始消费，中间未处理的数据就会丢失。另外，手动指定group.id在Spark结构化流中容易引发offset管理冲突，因为Spark本身有自己的offset追踪机制。

具体解决步骤： #

1. 关闭Kafka自动提交，改用Spark Checkpoint管理offset
   把enable.auto.commit设为false，同时在写入流时指定checkpointLocation。Spark会将offset和作业状态持久化到checkpoint目录，重启后自动从上次中断的位置继续消费，彻底避免数据丢失。

2. 开启优雅停止，确保当前批次处理完成
   添加Spark配置spark.streaming.stopGracefullyOnShutdown=true，当收到停止信号时，Spark会等待当前正在处理的批次完成后再停止，不会中途丢弃数据。

3. 移除手动指定的group.id（可选但推荐）
   Spark结构化流会自动为每个作业生成唯一的消费组ID，手动指定group.id可能和Spark的offset管理机制冲突，导致offset追踪混乱。移除后由Spark自动管理更可靠。

修改后的代码示例： #

// 先设置Spark优雅停止的配置
spark.conf.set("spark.streaming.stopGracefullyOnShutdown", "true")

val df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "x.x.x.x")
  .option("subscribe", "topic-name")
  .option("enable.auto.commit", "false") // 关闭自动提交
  // .option("kafka.group.id","group_name") // 建议移除，由Spark自动管理
  .load()

// 写入流时指定checkpoint目录
val query = df.writeStream
  .format("console") // 替换成你的实际输出格式（比如parquet、jdbc等）
  .option("checkpointLocation", "/path/to/your/checkpoint/dir") // 必须是持久化存储目录（如HDFS、本地稳定目录）
  .start()

query.awaitTermination()

额外注意事项： #

• checkpoint目录必须用持久化存储，不能用临时目录，否则重启后会丢失作业状态。
• 如果必须保留手动指定的group.id，要确保同一消费组下只有一个Spark作业在运行，避免多个消费者引发重复消费或offset冲突。

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