Spark批处理从Kafka读取（kafka-2.1.1-0.9.0.1）：如何获取until offset？

嘿，针对你用的Kafka 2.1.1-0.9.0.1版本和Spark Streaming的旧Kafka API，我给你两种常见场景下获取until offset的实用方案：

场景1：读到当前Kafka分区的最新偏移量 #

如果你的需求是从已获取的from offset读到当前分区的最新位置，直接复用你已经在用的KafkaCluster.getLatestLeaderOffsets方法就可以了。这个方法返回的就是每个分区当前的最新偏移量（也就是该分区最后一条消息的下一个偏移位置，完全符合Kafka消费偏移量的定义）。

示例代码片段：

import org.apache.spark.streaming.kafka.KafkaCluster
import kafka.common.TopicAndPartition

// 假设你已经配置好Kafka参数并初始化了KafkaCluster实例
val kafkaCluster = new KafkaCluster(kafkaParams)
val targetTopic = "your_topic_name"

// 获取目标topic的所有分区
val partitionsOpt = kafkaCluster.getPartitions(Set(targetTopic))
if (partitionsOpt.isRight) {
  val partitions = partitionsOpt.right.get
  // 获取最新偏移量作为until offset
  val untilOffsetsOpt = kafkaCluster.getLatestLeaderOffsets(partitions)
  if (untilOffsetsOpt.isRight) {
    val untilOffsets = untilOffsetsOpt.right.get.map { case (tp, leaderOffset) =>
      tp -> leaderOffset.offset
    }.toMap
    // 现在untilOffsets就是每个分区的结束偏移量，可直接用于createRDD
  }
}

场景2：读到指定时间点对应的偏移量 #

如果你的需求是读取某个时间范围内的消息（比如从from offset读到某个时间点之前的所有消息），由于Kafka 0.9的消费者API没有直接提供「时间转偏移量」的方法，你需要用Kafka的SimpleConsumer来手动发送偏移量请求：

具体步骤如下：

• 先通过ZooKeeper获取目标topic的所有分区及其对应的leader节点
• 针对每个分区，用SimpleConsumer向leader发送OffsetRequest，指定目标时间戳
• 解析响应得到对应的偏移量作为until offset

示例代码片段：

import kafka.api.{OffsetRequest, OffsetResponse}
import kafka.common.TopicAndPartition
import kafka.consumer.SimpleConsumer
import kafka.utils.ZkUtils

// 配置ZooKeeper地址、目标topic和要查询的时间戳（毫秒）
val zkQuorum = "your_zk_host:2181"
val targetTopic = "your_topic_name"
val targetTimestamp = 1620000000000L // 替换成你需要的时间戳

// 初始化ZkUtils获取分区和leader信息
val zkUtils = ZkUtils(zkQuorum, 30000, 30000, false)
val partitions = zkUtils.getPartitionsForTopics(Set(targetTopic)).get(targetTopic).get
val partitionLeaderMap = partitions.map { partition =>
  val tp = TopicAndPartition(targetTopic, partition)
  val leaderOpt = zkUtils.getLeaderForPartition(targetTopic, partition)
  tp -> leaderOpt.get // 实际开发中要处理leader不存在的异常情况
}.toMap

// 遍历每个分区，查询对应时间戳的偏移量
val untilOffsets = partitionLeaderMap.map { case (tp, leader) =>
  val consumer = new SimpleConsumer(leader.host, leader.port, 30000, 1024*1024, "offset-lookup-consumer")
  // 构造OffsetRequest：指定目标时间戳，最多返回1个偏移量
  val offsetRequest = OffsetRequest(Map(tp -> OffsetRequest.PartitionRequest(targetTimestamp, 1)))
  val offsetResponse = consumer.getOffsetsBefore(offsetRequest)
  consumer.close()
  // 取返回的第一个偏移量作为该分区的until offset
  tp -> offsetResponse.partitionErrorAndOffsets(tp).offsets.head
}.toMap

小提示：如果把targetTimestamp换成-1，会返回该分区的最新偏移量；换成-2则返回最早偏移量，和getLatestLeaderOffsets的效果一致。

一些注意事项 #

• 如果你用getLatestLeaderOffsets获取until offset，要留意时间差：如果获取from offset和until offset的间隔较长，期间可能有新消息写入，这是批处理场景下的正常情况。
• 使用SimpleConsumer时，一定要处理leader节点不存在或者请求失败的异常，避免程序崩溃。
• 这些方案完全兼容你在用的旧版Spark Kafka API，拿到fromOffsets和untilOffsets后，就可以直接构造KafkaUtils.createRDD需要的偏移范围了。

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