Spark Structured Streaming读取Kafka时offset始终从最早开始的问题

问题分析 #

我碰到过不少Spark 2.2.1版本对接Kafka Structured Streaming时的这个坑，刚好能帮你理清问题：这个版本里，当你的流查询首次启动（无checkpoint记录），或者checkpoint里没有保存对应消费者组的偏移量时，你配置的auto.offset.reset=latest并不会生效，框架会默认强制用earliest作为初始偏移量——这是该版本的特定行为，和Kafka新消费者API的常规逻辑有差异。

解决方案 #

针对你要从最新偏移量开始读取的需求，分两种场景处理：

场景1：首次启动流查询（无checkpoint） #

必须显式通过startingOffsets参数指定初始偏移量为latest，不能只依赖auto.offset.reset。示例代码如下：

Scala API 示例

val kafkaStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-broker-list:9092")
  .option("subscribe", "your-target-topic")
  .option("startingOffsets", "latest") // 核心配置：强制初始从最新偏移量启动
  .option("auto.offset.reset", "latest") // 兜底：后续若偏移量丢失时用latest策略
  .load()

SQL 方式示例

如果用SQL语法创建流表，同样要指定startingOffsets：

CREATE STREAMING TABLE kafka_topic_data
USING kafka
OPTIONS (
  kafka.bootstrap.servers = 'your-broker-list:9092',
  subscribe = 'your-target-topic',
  startingOffsets = 'latest',
  auto.offset.reset = 'latest'
);

场景2：已有checkpoint但需要重置到最新偏移量 #

如果你的流查询已经运行过，checkpoint目录里保存了旧的偏移量记录，需要先删除现有的checkpoint目录，再按照场景1的配置重新启动查询，这样才能触发从最新偏移量开始读取的逻辑。

额外说明 #

这个问题在Spark 2.3及以上版本已经被修复，auto.offset.reset的配置会在初始启动时正常生效，但由于你使用的是2.2.1版本，必须通过startingOffsets强制指定初始偏移量才能达到预期效果。

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