卡夫卡聚合流數據流

我生產的消息卡夫卡有了這樣的代碼：卡夫卡聚合流數據流

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))); 
}

我想算與卡夫卡流最後一小時（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");

我對卡夫卡/溪流是如此的陌生。我該怎麼做？

來源

2016-10-19 kamaci

這將是有益的，如果你可以給一個數據例如，輸入，輸出和預期輸出。代碼看起來很好。不確定你面臨的問題是什麼。 –

另外，您面臨的錯誤/問題是什麼？在你的問題中沒有提到。我看到你分享的一小部分代碼存在一些問題，但在跳到過早的結論之前，如Matthias J. Sax所說，如果你可以分享更多的信息 - 特別是關於Kafka Streams相關代碼，它將會有所幫助： 1.您已經定義了與Kafka Streams相關的配置設置，以及2.完整的拓撲定義等，包括您通過''KafkaStreams＃start（）''實際啓動拓撲的部分。 –

首先..你缺乏這種代碼真正開始您的流處理..

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

來源

2017-05-18 12:43:12

卡夫卡聚合流數據流

回答

相關問題