1

次のようなコードで 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 Streams (0.10.0.1) で過去 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が初めてです。どうすればいいですか？

4

3 に答える 3

1

2 つのストリームを集約するには、join メソッドを利用できます。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();
于 2018-08-22T01:01:14.637 に答える