apache-kafka - 在使用 kafka-python 时定期轮询 Kafka 消费者的最佳方法是什么？

Question

我有多个生产者正在向 Kafka 提供数据。我希望每小时运行一个消费者以一次获取所有累积的数据并进一步处理它。

我想到的选项是:

使用 python 线程并使用等效的 setInterval 来调用消费者

设置 max_poll_interval_ms 变量:(如其他一些答案中所述)。但是，官方文档指出

This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance This does not sound like that it's responsible for putting a consumer to sleep and then firing it again.

我不是每小时轮询一次，而是跟踪 Consumer Offset 并在 10,000 条记录附加到 Kafka 后进行轮询

但是，我想在消费者本身内管理相同的内容。最好的方法是什么
?

Answer 1

使用 Cron 或您的操作系统调度程序每小时调用一个脚本。

如果您需要等到 10k 条记录出现在该主题上才能做任何有用的事情，那么我并不完全确定 Kafka 是否适合该架构。此外，消费者滞后实际上会不断落后