Spark Streaming作业重启后从Kafka开头消费无法断点续传问题

问题背景 #

在Google Cloud Platform上使用Dataproc运行Spark Structured Streaming作业，以Kafka为数据源，最终将处理后的数据写入MongoDB。作业运行期间状态正常，但故障重启后会从Kafka topic的起始位置重新消费消息，无法从上次停止的偏移量位置续传。

现有代码配置 #

Kafka 数据源读取配置 #

clickstreamTestDf = (
  spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", confluentBootstrapServers)
  .option("kafka.security.protocol", "SASL_SSL")
  .option("kafka.sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='{}' password='{}';".format(confluentApiKey, confluentSecret))
  .option("kafka.ssl.endpoint.identification.algorithm", "https")
  .option("kafka.sasl.mechanism", "PLAIN")
  .option("subscribe", "customer_experience")
  .option("failOnDataLoss", "false")
  .option("startingOffsets", "earliest")
  .load()
)

MongoDB 写入流配置 #

finished_df.writeStream \
    .format("mongodb")\
    .option("spark.mongodb.connection.uri", connectionString) \
    .option("spark.mongodb.database", "Company-Environment") \
    .option("spark.mongodb.collection", "customer_experience") \
    .option("checkpointLocation", "gs://firstsparktest_1/checkpointCustExp") \
    .option("forceDeleteTempCheckpointLocation", "true") \
    .outputMode("append") \
    .start() \
    .awaitTermination()

待解答疑问 #

• 是否需要将startingOffsets参数设置为latest？此前尝试过该配置，但作业仍未从上次停止的位置续传消费。
• 当前checkpointLocation的配置方式是否正确？使用Google Storage目录作为检查点存储路径是否可行？
• 目标场景：运行流作业后手动停止作业、删除Dataproc集群，次日新建集群提交作业时可从上次停止位置继续消费，该需求是否可实现？具体需要如何配置？

解决方案 #

1. startingOffsets参数的作用范围 #

startingOffsets仅在作业首次启动、指定检查点路径下不存在有效已提交偏移量记录时生效。一旦检查点目录中存储了作业运行时提交的消费偏移量，后续不管是故障重启还是手动重启作业，Spark都会直接读取检查点内记录的偏移量续传，完全忽略该参数的配置值。
此前修改该参数为latest仍无法续传，和参数本身无关，核心原因是检查点没有被正确持久化，或者重启时检查点已经被删除，作业只能 fallback 到startingOffsets指定的初始消费位置。

2. 检查点配置问题说明 #

Google Storage（GCS）完全可以作为Spark Structured Streaming的检查点存储路径，Dataproc集群默认集成了GCS对应的Hadoop FileSystem实现，不需要额外引入依赖或者做特殊配置即可正常读写gs://路径下的检查点文件。
当前配置失效的核心原因是写入流中设置了.option("forceDeleteTempCheckpointLocation", "true")，该参数会在作业停止时自动清空整个检查点目录，重启作业时找不到之前存储的偏移量和作业状态，自然会从头开始消费，直接删除该参数配置即可。
另外需要确认权限配置：运行作业的Dataproc服务账号需要对检查点对应的GCS路径拥有读写权限，作业运行期间不要手动修改、删除检查点目录下的任何文件，也不要在多个独立的流作业之间复用同一个检查点路径。

3. 跨集群续传消费的实现方法 #

删除集群后新建集群续传消费的需求完全可以实现，只需要满足以下配置要求：

• 持久化保留检查点目录：删除forceDeleteTempCheckpointLocation配置，停止作业、删除Dataproc集群时不要手动清理GCS上对应的检查点目录，该目录存储的消费偏移量、作业状态是实现续传的核心依据。
• 新集群提交作业时，必须使用和之前作业完全一致的检查点路径，同时流处理的核心逻辑（比如数据源配置、schema定义、sink配置、核心算子链路）不能做不兼容修改，否则Spark会因为检查点记录的状态和当前作业逻辑不匹配抛出异常。
• 确认Kafka侧的日志保留时长覆盖停机窗口：如果停机时间超过Kafka topic配置的日志保留周期，之前记录的偏移量对应的消息已经被Kafka清理，作业会按照failOnDataLoss=false的配置自动跳到当前topic现存的最早偏移量继续消费，不会直接中断作业。
• 首次启动作业时startingOffsets可按需配置：设为earliest即首次启动从头消费全量历史数据，设为latest即首次启动从topic最新位置开始消费，后续重启不管集群是否重建，只要检查点目录完整有效，就会从上次作业停止时已提交的偏移量位置继续消费。

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