Spark 2.4下Kafka关闭enable.auto.commit时offset处理及API差异问题

前置场景：Kafka enable.auto.commit 参数设置为 false，Spark版本为2.4

1. latest offset的读取逻辑与手动指定场景 #

• 默认不需要手动查询最新offset传入KafkaUtils.createDirectStream()，框架会自动读取Kafka broker对应分区的最新offset。
• 需要手动获取并指定latest offset的场景：
  • 业务需要跳过中间段异常数据，直接从最新位置消费，不走默认的已提交offset位点
  • 要求应用每次重启都强制从最新位点消费，不受历史提交offset影响
  • 自定义offset管理逻辑时，需要先拉取latest offset做范围校验、消费边界限制等操作

2. SparkSession.readStream.format("kafka")与KafkaUtils.createDirectStream()的差异 #

• 所属API体系不同：
  • SparkSession.readStream.format("kafka")是Structured Streaming高阶API，为Spark 2.x之后官方主推的流处理方案，基于DataFrame/DataSet做数据抽象
  • KafkaUtils.createDirectStream()是Spark Streaming（DStream）低阶API，属于Spark早期的流处理实现
• offset管理逻辑不同：
  • Structured Streaming的Kafka源默认将offset维护在checkpoint目录，开启checkpoint即可保证恰好一次语义，不需要手动操作offset
  • DStream的createDirectStream()在enable.auto.commit=false的前提下，需要自行实现offset的存储、提交逻辑，提交时机完全由业务控制
• 功能支持不同：
  • Structured Streaming天然支持事件时间、窗口计算、水印等复杂流处理特性，与Spark SQL能力完全打通，开发效率更高
  • DStream基于批次时间做流处理，事件时间等复杂逻辑需要手动实现，开发成本更高
• 迭代状态不同：Structured Streaming为官方重点迭代方向，新特性优先支持；Spark 3.x之后DStream API已经停止新功能开发，仅做bug修复。

3. earliest offset的处理逻辑 #

框架会自动识别earliest配置，但有触发限制：

• 对于KafkaUtils.createDirectStream()：只有没有手动指定起始offset、也没有查询到对应消费组的已提交offset时，earliest配置才会生效，框架会从分区最早可用offset开始消费；如果存在已提交的offset，默认优先从已提交位置消费，要强制走earliest需要手动指定起始offset。
• 对于Structured Streaming的Kafka源：配置startingOffsets = "earliest"仅在应用首次启动、checkpoint目录为空时生效，后续重启会优先读取checkpoint中存储的offset，不会再触发earliest逻辑，要强制重置需要清空checkpoint目录。

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