如何在Spark Streaming Scala中从指定偏移量消费Kafka消息

方法一：使用Assign策略手动指定偏移量 #

直接用Assign替代Subscribe策略，明确指定要消费的分区及对应起始偏移量，这是最直接的实现方式：

1. 先获取目标topic的分区信息（可通过Kafka API或命令行工具获取）
2. 构建分区与偏移量的映射关系
3. 基于Assign策略创建DirectStream

修改后的Scala代码示例：

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import io.confluent.kafka.serializers.KafkaAvroDeserializer

val conf = new SparkConf().setMaster("local[2]").setAppName("KafkaOffsetReset")
val ssc = new StreamingContext(conf, Seconds(1))

val topic = "test123"
val kafkaParams = Map(
  "bootstrap.servers" -> "localhost:9092",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[KafkaAvroDeserializer],
  "schema.registry.url" -> "http://abc.test.com:8089",
  "group.id" -> "spark-streaming-notes",
  "enable.auto.commit" -> false // 建议关闭自动提交，手动管理偏移量
)

// 指定要消费的分区和对应的起始偏移量（假设test123只有一个分区0，起始偏移量1020）
val partition0 = new TopicPartition(topic, 0)
val offsets = Map(partition0 -> 1020L)

// 使用Assign策略创建流
val stream = KafkaUtils.createDirectStream[String, Object](
  ssc,
  LocationStrategies.PreferConsistent,
  ConsumerStrategies.Assign[String, Object](offsets.keys.toList, kafkaParams, offsets)
)

// 处理消息，添加业务逻辑（比如写入数据库）
stream.foreachRDD { rdd =>
  // 消息处理完成后手动提交偏移量
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}

stream.print()

ssc.start()
ssc.awaitTermination()

注意：如果topic存在多个分区，需要为每个分区都指定对应的目标偏移量。

方法二：通过Kafka命令行重置消费者组偏移量 #

如果不想修改代码，可先在Kafka服务器上用kafka-consumer-groups.sh脚本直接重置目标消费者组的偏移量，再启动Spark作业：

# 查看消费者组当前的偏移量状态
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group spark-streaming-notes

# 重置指定topic分区的偏移量到1020（假设分区为0）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --to-offset 1020 --topic test123:0 --execute --group spark-streaming-notes

执行完重置命令后，启动你的Spark Streaming作业，即可从指定的1020偏移量开始消费。

额外建议 #

• 关闭enable.auto.commit，改为手动提交偏移量，确保只有当消息成功写入数据库等存储后再提交偏移量，避免消息丢失。
• 若需要频繁重置偏移量，建议维护一个外部存储（如MySQL、Redis）来管理偏移量，作业启动时从外部存储读取指定偏移量，实现更灵活的偏移量控制。

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