如何使用 confluent-kafka-python 消费过去 N 天的消息？答案

【问题标题】：How to consume messages in last N days using confluent-kafka-python?如何使用 confluent-kafka-python 消费过去 N 天的消息？
【发布时间】：2022-02-21 05:19:40
【问题描述】：

这个问题类似于Python KafkaConsumer start consuming messages from a timestamp，除了我想知道如何在 Confluent 的官方 Python Kafka 客户端中做到这一点。

我查看了 Consumer.offsets_for_times 函数，但我对它接受 TopicPartition.offset 字段中的时间戳感到困惑。

offset 如何等同于时间戳？

【问题讨论】：

标签： python apache-kafka confluent-kafka-python

【解决方案1】：

我最近这样做是为了 $work。你需要得到offsets_for_times()的结果，然后assign()那个列表给你的消费者，然后调用consume()。重要的是，不要 subscribe() 讨论该主题。（参见 Eden Hill 对https://github.com/confluentinc/confluent-kafka-python/issues/373 的评论）。

你说得对，这个函数的文档在定义时间戳和偏移量时有些混乱。

更新以回答后续问题：

与How do I get the the offset of last message of a Kafka topic using confluent-kafka-python? 的区别在于，而不是

topicparts = [TopicPartition(topic_name, i) for i in range(0, 8)]

你会做这样的事情：

whents = datetime.fromisoformat("2022-01-01T12:34:56.000")
whenms = int(whents) * 1000   # to get milliseconds

topicparts = [TopicPartition(topic_name, i, whenms) for i in range(0, 8)]

【讨论】：

感谢分享信息。如何为offsets_for_times() 定义TopicPartition 参数？你能详细说明吗？谢谢
我发现使用时间戳作为偏移量实际上是在 2016 年作为KIP-79 的一部分实现的。这是一个实际功能。谢谢。

【解决方案2】：

该方法不接受时间戳；只有您想要为其查找时间戳的分区。

https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#confluent_kafka.TopicPartition.TopicPartition

也许你的意思是 timeout 参数？

【讨论】：

这正是我的观点。这是我链接的文档中写的内容，令人困惑。
我不明白你认为什么令人困惑。您定义一个 TopicPartition 对象列表，它会返回每个对象的时间戳。您将为偏移参数提供OFFSET_STORED 值
不，该功能完全相反。它将“按时间戳查找指定分区的偏移量。”。阅读文档。