Spark批处理读取Kafka时借助Kafka跟踪消费偏移量的异常排查

• Spark 批处理模式的 Kafka 数据源，从设计上就不会自动读取指定kafka.group.id下存储的已提交offset作为读取起始点。哪怕你通过原生KafkaConsumer把offset提交到了对应消费组下，Spark启动批作业时也不会主动查询这些offset。
• 批模式下Spark Kafka数据源的默认起始读取规则为：未显式配置startingOffsets参数时，默认值是earliest，即从每个分区的最早可用offset开始全量读取，这就是你每次运行都从头消费的直接原因。
• 你之前配置的kafka.group.id在Spark批读取过程中，仅作为参数透传存在，既不会被用来查询已提交offset，也不会被用来关联offset提交逻辑，等同于一个自定义标识，没有实际的消费状态同步作用。
• 该行为和EmbeddedKafka环境无关，哪怕在生产级Kafka集群上，只要遵循同样的调用逻辑，结果完全一致。
• 注意：该逻辑和Structured Streaming的Kafka数据源完全不同，流处理场景下Spark会自动维护消费状态、甚至支持将offset提交到Kafka消费组，但批处理Kafka数据源从设计之初就没有实现这部分自动状态同步的能力。

要实现每日跑批、按消费组记录的offset范围读取增量数据的目标，需要手动完成「拉取已提交offset-传递给Spark读取边界-作业成功后提交新offset」的全链路逻辑，步骤如下：

1. 作业启动阶段主动拉取消费组的已提交offset
   初始化原生KafkaConsumer，指定和Spark作业相同的group.id，查询目标topic下所有分区对应的已提交offset。同时加兜底逻辑：对于第一次运行、或者Kafka扩容后新增的分区，如果不存在已提交的offset，默认设置从分区最早位置开始读取。
   示例代码：

   val topicName = "topic-in"
   val groupId = "consumer-group-for-this-job"
   val kafkaBrokers = "localhost:6001"
   
   val consumerProps = new Properties()
   consumerProps.put("bootstrap.servers", kafkaBrokers)
   consumerProps.put("group.id", groupId)
   consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")
   consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")
   val kafkaConsumer = new KafkaConsumer[Array[Byte], Array[Byte]](consumerProps)
   
   // 获取目标topic的全量分区列表，兼容后续分区扩容的场景
   val topicPartitions = kafkaConsumer.partitionsFor(topicName).asScala
     .map(p => new TopicPartition(p.topic(), p.partition()))
     .asJavaCollection
   
   // 拉取当前消费组下每个分区的已提交offset
   val committedOffsets = kafkaConsumer.committed(topicPartitions).asScala
   // 构建起始offset映射，无已提交offset的分区默认从0位置开始
   val startingOffsets = topicPartitions.asScala.map { tp =>
     val offsetMeta = committedOffsets.get(tp)
     val startOffset = if (offsetMeta == null) 0L else offsetMeta.offset()
     tp -> startOffset
   }.toMap
   

2. 将拉取到的offset作为读取边界传给Spark
   把查询到的起始offset按照Spark Kafka数据源要求的JSON格式，配置到startingOffsets参数中，明确告诉Spark本次作业需要从哪个位置开始读取。手动指定的边界优先级最高，不需要依赖Spark自动识别消费组的offset。
   示例代码：

   // 转换为Spark Kafka数据源要求的json格式起始offset
   val startingOffsetsJson = Utils.json(startingOffsets.map {
     case (tp, offset) => tp.topic -> Map(tp.partition.toString -> offset)
   })
   
   val etlData = spark.read
     .format("kafka")
     .option("kafka.bootstrap.servers", kafkaBrokers)
     .option("subscribe", topicName)
     .option("includeHeaders", "true")
     .option("startingOffsets", startingOffsetsJson)
     // 结束点默认取作业启动时的最新offset，也可以根据需求手动配置
     .load()
   

3. 作业全流程成功后再提交offset
   等Spark作业的所有核心处理、下游写入步骤全部执行完成、无报错后，再收集本次作业实际处理到的每个分区的最大offset。使用原生KafkaConsumer把这些offset提交到指定消费组下，为下次跑批做准备。
   示例代码：

   // 所有下游写入操作执行完成、作业无报错后，收集本次处理的最大offset
   val processedOffsets = etlData
     .select("topic", "partition", "offset")
     .groupBy("topic", "partition")
     .agg(max("offset").as("max_offset"))
     .as[(String, Int, Long)]
     .collect()
     .map { case (topic, partition, maxOffset) =>
       // Kafka提交offset的语义是"下一条要消费的位置"，需要在最大已处理offset基础上+1
       new TopicPartition(topic, partition) -> new OffsetAndMetadata(maxOffset + 1)
     }.toMap.asJava
   
   // 同步提交offset
   kafkaConsumer.commitSync(processedOffsets)
   

• offset提交时机：你之前的代码在计算最大offset时就直接提交，此时Spark的核心ETL逻辑根本没有实际执行（DataFrame是懒加载机制），相当于在作业还没跑的时候就告知Kafka"我已经处理完成"，会直接导致数据丢失。必须等所有写入操作完成、作业整体成功后再提交。
• offset off-by-one问题：Kafka消费组提交offset的语义是「下一条需要消费的消息的位置」。如果作业处理到分区的最大offset为100，需要提交的offset值应该是101，否则下次运行时会从100的位置开始，重复读取上一次已经处理过的最后一条消息，造成数据重复。
• 新分区兼容：如果Kafka topic后续做了分区扩容，拉取offset时要通过partitionsFor方法动态获取全量最新的分区列表，不要硬编码固定的分区数，避免新分区的数据被漏读。

• 大部分场景下不需要额外把offset存储到HDFS、关系库、缓存等外部系统，直接使用Kafka自带的消费组offset存储就足够满足每日跑批的状态跟踪需求。
• 如果你有极高的可靠性要求，担心Kafka集群的offset数据被误删、重置，可以在提交Kafka offset的同时，额外将offset持久化到外部存储，作为兜底恢复的备份，但这属于额外的加固逻辑，不是必需项。

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