Блог

У меня возникла проблема с kafka consumer в python, вот мой вариант использования, kafka producer будет отправлять данные каждую секунду в потоковом режиме реального времени. Но потребителю kafka необходимо использовать каждые 30 минут и собирать пакет данных. Помогите мне решить.

Если вы не хотите обрабатывать свои данные в режиме реального времени, вы можете пересмотреть, является ли 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 минут.