Блог

Главная - Вопросы по программированию - Агрегировать потоковые данные с потоками Kafka

Агрегировать потоковые данные с потоками Kafka

#apache-kafka #apache-kafka-streams

#apache-kafka #apache-kafka-streams

Вопрос:

Я создаю сообщения для Kafka с помощью такого кода:

 Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testo");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

Producer<String, String> producer = new KafkaProducer<>(props);

for (int i = 0; i < 1000; i  ) {
  producer.send(new ProducerRecord<>(
    "topico",
    String.format("{"type":"test", "t":%.3f, "k":%d}", System.nanoTime() * 1e-9, i)));
}

Я хочу подсчитать общее количество сообщений за последний час с потоками Kafka (0.10.0.1). Я пробовал это:

 final KStreamBuilder builder = new KStreamBuilder();
final KStream<String, String> metrics = builder.stream(Serdes.String(), Serdes.String(), "topico");
metrics.countByKey(TimeWindows.of("Hourly", 3600 * 1000)).mapValues(Object::toString).to("output");

Я новичок в Kafka / Streams. Как я могу это сделать?

Комментарии:

1. Было бы полезно, если бы вы могли привести пример данных с вводом, выводом и ожидаемым результатом. Код кажется прекрасным. Не уверен, с какими проблемами вы сталкиваетесь.

2. Кроме того, с какой ошибкой / проблемой вы столкнулись? Это не упоминается в вашем вопросе. Я вижу пару проблем с небольшим фрагментом кода, которым вы поделились, но прежде чем делать преждевременные выводы, было бы полезно, как сказал Маттиас Дж. Сакс, если бы вы могли поделиться дополнительной информацией — в частности, о коде, связанном с потоками Kafka: 1. параметры конфигурации, которые вы определили, относящиеся к потокам Kafka, и 2. полное определение топологии и т.д., Включая ту часть, где вы фактически запускаете топологию через sth, например KafkaStreams#start() .

Ответ №1:

Прежде всего.. Вам не хватает этого кода, чтобы фактически запустить процесс потоковой передачи.. 

 KafkaStreams streams = new KafkaStreams(builder, config);   
streams.start();    
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

Ответ №2:

Для агрегирования двух потоков вы можете использовать метод объединения. В kstreams доступны различные объединения.

Например: если вы хотите присоединиться kstream с ktable : 

 KStream<String, String> left = builder.stream("topic1");
KTable<String, String> right = builder.table("topic2");

left.leftjoin((right, (leftValue, rightValue) -> Customfunction(rightValue, leftValue))

наконец, запустите kstream 

 streams = new KafkaStreams(topology, config);
streams.start();

Ответ №3:

Я также новичок в потоках kafka, я не знаю старого API, но с новым (2.1.x) должно сработать что-то вроде этого 

  kstream.mapValues((readOnlyKey, value) -> "test")
                    .groupByKey()
                    .windowedBy(TimeWindows.of(1000 * 60))
                    .count()
                    .toStream()
                    .selectKey((key, value) -> Instant.ofEpochMilli(key.window().end())
                            .truncatedTo(ChronoUnit.HOURS).toEpochMilli())
                    .groupByKey(Serialized.with(Serdes.Long(), Serdes.Long())).reduce((reduce, newVal) -> reduce   newVal)
                    .toStream().peek((key, value) -> log.info("{}={}",key,value));