Apache Kafka最旧记录读取方法及SparkStreaming消费故障恢复问询

一、如何读取Apache Kafka中的最旧记录？ #

分两种场景给你说明：

• 如果是新的消费者组：直接在消费者配置里把auto.offset.reset设为earliest，启动后就会自动从主题的第一条记录开始消费。
• 如果是已经消费过的旧消费者组：因为Kafka会保存该组的消费偏移量，直接改配置没用，得手动重置偏移量。用Kafka自带的命令行工具就行：

  kafka-consumer-groups.sh --bootstrap-server <你的Kafka broker地址> --group <你的消费者组ID> --reset-offsets --to-earliest --topic <目标主题名> --execute
  

  执行完这个命令，下次启动该消费者组就会从头拉取数据了。

二、Spark Streaming消费的断点续传与最坏情况从头消费 #

你猜的没错！checkpointing确实是实现断点续传的核心手段，但要配合正确的配置才能保证绝对不丢数据，我给你详细拆解：

1. 实现断点续传（重启后从上次中断处继续） #

要达成这个目标，得同时做好这几点：

• 开启Spark Checkpoint：创建StreamingContext的时候指定一个持久化的checkpoint目录（生产环境建议用分布式存储比如HDFS）。代码大概是这样：

  // 先定义创建流处理逻辑的方法
  def createStreamingContext(): StreamingContext = {
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(5))
    // 这里写你的Kafka消费和处理逻辑
    ssc.checkpoint(checkpointDir)
    ssc
  }
  // 从checkpoint恢复，没有的话就创建新的
  val ssc = StreamingContext.getOrCreate(checkpointDir, createStreamingContext _)
  

  这样Spark会把消费偏移量和流的状态存在checkpoint里，重启时自动恢复到上次中断的位置。
• 用DirectStream消费：别用Receiver方式，DirectStream会直接从Kafka brokers拉数据，偏移量直接由Spark管理（存在checkpoint里），而且默认是处理完一个批次后才提交偏移量，能保证数据不丢。
• 保证下游写入的可靠性：你说不担心重复，那只要确保数据成功写入下游（比如数据库、HDFS）后，Spark才提交偏移量（DirectStream默认就是这样），就不会出现丢失的情况。

2. 最坏情况下从最旧记录开始消费 #

如果遇到checkpoint损坏、偏移量丢失这种极端情况，想要从头消费，有几种办法：

• 删除checkpoint目录：如果是用checkpoint管理偏移量，直接删掉对应的目录，然后重启任务，同时确保你的Kafka配置里auto.offset.reset=earliest，这样Spark会以新的状态从头开始消费。
• 重置消费者组偏移量：和第一个问题里的方法一样，用Kafka命令行工具把Spark Streaming用的消费者组偏移量重置到earliest，重启任务后就会从头拉取。
• 代码里强制指定起始偏移量：如果是用DirectStream，可以手动指定每个分区的起始偏移量为0（也就是最旧的位置），代码示例：

  // 先获取主题的所有分区
  val kafkaAdminClient = AdminClient.create(kafkaParams)
  val topicPartitions = kafkaAdminClient.describeTopics(Seq(topicName)).values().get(topicName).get().partitions().asScala
    .map(p => new TopicPartition(topicName, p.partition()))
  // 手动设置每个分区从0开始消费
  val fromOffsets = topicPartitions.map(tp => tp -> 0L).toMap
  // 创建DirectStream时指定偏移量
  val kafkaStream = KafkaUtils.createDirectStream[String, String](
    ssc,
    PreferConsistent,
    Assign[String, String](fromOffsets.keys, kafkaParams, fromOffsets)
  )
  

  这种方式不管之前的偏移量情况，强制从头开始消费。

另外补充一点：为了绝对不丢数据，还要确保Kafka主题的retention.ms设置足够大，保证消费者故障期间，未消费的记录不会被Kafka自动删除。

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