如何在python中每30分钟消耗一次kafka队列答案

【问题标题】：How to consume kafka queue every 30 minute in python如何在python中每30分钟消耗一次kafka队列
【发布时间】：2019-03-15 10:03:09
【问题描述】：

我在 python 中遇到了关于 kafka 消费者的问题，这是我的用例 kafka 生产者将在实时流中每秒发送数据。但是kafka消费者需要每30分钟消费一次，收集一批数据。帮我解决。

【问题讨论】：

标签： apache-kafka kafka-consumer-api kafka-python

【解决方案1】：

如果您不想实时处理数据，您可能需要重新考虑 Kafka 是否适合您。不过，你可以试试这个：

 Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("group.id", "your_consumer_group");
 props.put("enable.auto.commit", "true");
 props.put("auto.commit.interval.ms", "1000");
 props.put("session.timeout.ms", "30000");
 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
 consumer.subscribe(Arrays.asList("foo", "bar"));
 while (true) {
     ConsumerRecords<String, String> records = consumer.poll(1000);
     for (ConsumerRecord<String, String> record : records) {
         System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
     }

     // After data is consumed, you make your thread sleep until next 30 min:
     Thread.sleep(30 * 60 * 1000);
 }

如果您希望在每小时的第 30 分钟或第 0 分钟执行一次实时批处理，则可以改用此睡眠：

Thread.sleep(System.currentTimeMillis() % (30*60*1000));

它会让您的消费者在00:00、00:30、01:00、01:30 等处醒来。点此链接了解详情：https://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html

同样，您可能不想以这种方式使用 kafka。最好将数据转储到某个存储（例如，按日期时间分区的 parquet 文件）并每 30 分钟对其进行一次批处理。

【讨论】：