使用 Kafka Streams 聚合流数据答案

【问题标题】：Aggregate Stream Data with Kafka Streams使用 Kafka Streams 聚合流数据
【发布时间】：2019-01-26 08:29:31
【问题描述】：

我正在使用这样的代码向 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) 的消息总数。我试过了：

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 很陌生。我该怎么做？

【问题讨论】：

如果您能给出一个数据示例，包括输入、输出和预期输出，将会很有帮助。代码看起来不错。不确定您面临的问题是什么。
另外，您面临的错误/问题是什么？你的问题中没有提到。我确实看到您分享的一小段代码存在一些问题，但在过早得出结论之前，它会有所帮助，正如 Matthias J. Sax 所说，如果您可以分享更多信息 - 特别是关于 Kafka Streams 相关代码的信息： 1. 您定义的与 Kafka Streams 相关的配置设置，以及 2. 完整的拓扑定义等，包括您通过 KafkaStreams#start() 等实际启动拓扑的部分。

标签： apache-kafka apache-kafka-streams

【解决方案1】：

首先..您缺少此代码来实际启动流式处理过程..

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

【讨论】：

【解决方案2】：

要聚合两个流，您可以使用 join 方法。 kstreams 中有 different joins 可用。

例如：如果你想加入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));

【讨论】：