Я отправляю сообщения Кафке с таким кодом:
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 Streams (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. Как я могу это сделать?
KafkaStreams#start()
. - person Michael G. Noll   schedule 20.10.2016