如何从Kafka中的旧偏移点获取数据？

Question

我使用动物园管理员从Kafka获取数据。在这里，我总是从最后一个偏移点获取数据。有没有办法指定偏移时间来获取旧数据？

有一个选项“自动偏移”。重置。它接受最小的或最大的。有人能解释一下什么是最小的和最大的吗。可以自动偏移。重置有助于从旧偏移点而不是最新偏移点获取数据？

Answer 1

请参考关于kafka config：http://kafka.apache.org/08/configuration.html的文档，查询偏移参数的最小值和最大值。

顺便说一句，在探索Kafka时，我想知道如何为消费者重播所有消息。我的意思是，如果一个消费者团体已经调查了所有的信息，并且想重新获得这些信息。

它可以实现的方式是从动物园管理员删除数据。使用kafka.utils.ZkUtils类删除zoowatch上的节点。下面是它的用法：

ZkUtils.maybeDeletePath(${zkhost:zkport}", "/consumers/${group.id}");

Answer 2

从Kafka文档中，他们说“Kafka.api.OffsetRequest.EarliestTime（）在日志中找到数据的开头并从那里开始流式处理，Kafka.api.OffsetRequest.LatestTime（）将只流式处理新消息。不要假设偏移量0是开始偏移量，因为消息会随着时间的推移从日志中过时。”

在此处使用SimpleConsumerExample：https://cwiki.apache.org/confluence/display/KAFKA/0.8.0SimpleConsumer示例

类似的问题：Kafka高级消费者使用JavaAPI从主题获取所有消息（相当于--从头开始）

这可能会有帮助

Answer 3

使用者始终属于一个组，对于每个分区，Zookeeper跟踪分区中该使用者组的进度。

要从头开始获取，您可以删除Hussain引用的与进度相关的所有数据

ZkUtils.maybeDeletePath(${zkhost:zkport}", "/consumers/${group.id}");

您还可以指定所需分区的偏移量，如core/src/main/scala/kafka/tools/UpdateOffsetsInZK中所指定的。斯卡拉

ZkUtils.updatePersistentPath(zkClient, topicDirs.consumerOffsetDir + "/" + partition, offset.toString)

然而，偏移量没有时间索引，但您知道每个分区是一个序列。

如果您的消息包含一个时间戳（请注意，这个时间戳与Kafka收到您的消息的时刻无关），您可以尝试使用索引器，尝试通过将偏移量增加N来逐步检索一个条目，并将元组（主题X，第2部分，偏移量100，时间戳）存储在某处。

当您想从指定的时间时刻检索条目时，您可以对粗略索引应用二进制搜索，直到找到您想要的条目并从那里获取。