scala - Spark - 在不打开流的情况下获取 Kafka 的最早和最新偏移量

Question

我目前正在使用 spark-streaming-kafka-0-10_2.11将我的 spark 应用程序与 kafka 队列连接起来。对于 Streams，一切正常。但是对于特定场景，我只需要一次 kafka 队列的全部内容 - 为此我得到了更好地使用 KafkaUtils.createRDD 的建议。 ( SparkStreaming: Read Kafka Stream and provide it as RDD for further processing )

然而对于 spark-streaming-kafka-0-10_2.11我无法弄清楚如何为我的 Kafka 主题获取最早和最新的偏移量，这是创建偏移范围所需的我必须处理的 createRDD方法。

在不打开流的情况下获得这些偏移量的推荐方法是什么？任何帮助将不胜感激。

Answer 1

在阅读了几次讨论后，我能够从特定分区获得最早或最新的偏移量:

val consumer = new SimpleConsumer(host,port,timeout,bufferSize,"offsetfetcher");
val topicAndPartition = new TopicAndPartition(topic, initialPartition)
val request = OffsetRequest(Map(topicAndPartition -> PartitionOffsetRequestInfo(OffsetRequest.EarliestTime,1)))
val offsets = consumer.getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition).offsets

return offsets.head

但是，如何在 kafka_consumer.sh CLI 命令中复制“from_beginning”的行为是我不知道的 KafkaUtils.createRDD 方法。