时间戳

首页 > 解决方案 > Kafka 流和窗口化以在时间窗口内保持计数

java - Kafka 流和窗口化以在时间窗口内保持计数

问题描述

我是 Stackoverflow 的新手，如果问题被问得不好，请原谅我。非常感谢任何帮助/灵感！

我正在使用 Kafka 流将传入数据过滤到我的数据库中。传入的消息看起来像{"ID":"X","time":"HH:MM"}和其他一些参数，在这种情况下无关紧要。我设法让一个 java 应用程序运行，它从一个主题中读取并打印出传入的消息。现在我想做的是使用 KTables(?) 对具有相同 ID 的传入消息进行分组，然后使用会话窗口将表分组到时隙中。我想要在时间轴上连续运行 X 分钟的时间窗口。

首先当然是让 KTable 运行以计算具有相同 ID 的传入消息。我想做的应该是这样的：

ID       Count
X          1
Y          3
Z          1

不断更新，因此从表中删除具有过时时间戳的消息。

我不是百分百肯定，但我认为我想要的是 KTables 而不是 KStreams，对吗？如果这是实现我想要的结果的正确方法，我如何实现滑动窗口？

这是我现在使用的代码。它仅从主题中读取并打印传入的消息。

private static List<String> printEvent(String o) {
    System.out.println(o);
    return Arrays.asList(o);
}

final StreamsBuilder builder = new StreamsBuilder();
    builder.<String, String>stream(srcTopic)
    .flatMapValues(value -> printEvent(value));

我想知道我必须添加什么来实现我想要的上述输出，以及我把它放在我的代码中的什么位置。

在此先感谢您的帮助！

标签： javaapache-kafkaapache-kafka-streams

解决方案

是的，您需要 Ktable 和滑动窗口，我还建议您查看水印功能，以处理延迟传递消息。 例子

KTable<Windowed<Key>, Value> oneMinuteWindowed = yourKStream

.groupByKey()

.reduce(/*your adder*/, TimeWindows.of(60*1000, 60*1000), "store1m");
        //where your adder can be as simple as (val, agg) -> agg + val
        //for primitive types or as complex as you need

推荐阅读