来自不同Kafka主题的事件数的聚合(总和)

x4shl7ld 于 2021-06-05 发布在 Kafka

关注(0)|答案(1)|浏览(553)

我的应用程序有三个主题，它们接收属于用户的一些事件：

Event Type A -> Topic A
Event Type B -> Topic B
Event Type C -> Topic C

这将是消息流的一个示例：

Message(user 1 - event A - 2020-01-03) 
Message(user 2 - event A - 2020-01-03) 
Message(user 1 - event C - 2020-01-20)
Message(user 1 - event B - 2020-01-22)

我希望能够生成每个用户每月事件总数的报告，聚合三个主题中的所有事件，例如：

User 1 - 2020-01 -> 3 total events
User 2 - 2020-01 -> 1 total events

有三个kstream（每个主题一个），我如何每月执行此加法以获得三个不同主题的所有事件的总和？你能出示这个的代码吗？

apache-kafka apache-kafka-streams spring-cloud-stream spring-cloud-stream-binder-kafka

来源：https://stackoverflow.com/questions/59923658/aggregation-summation-of-number-of-events-from-different-kafka-topics

1条答案

按热度按时间

f0brbegy1#

因为您只对计数感兴趣，所以最简单的方法是只保留用户id作为键，并为每个用户保留一些伪值 KStream ，合并所有三个流，然后进行窗口计数（请注意，基于日历的窗口不支持开箱即用；您可以使用31天窗口作为近似值，也可以构建自己的自定义窗口）：

// just map to dummy empty string (note, that `null` would not work
KStream<UserId, String> streamA = builder.stream("topic-A").mapValues(v -> "");
KStream<UserId, String> streamB = builder.stream("topic-B").mapValues(v -> "");
KStream<UserId, String> streamC = builder.stream("topic-C").mapValues(v -> "");

streamA.merge(streamB).merge(streamC).groupByKey().windowBy(...).count();

你也可能对 suppress() 接线员。

赞(0）回复(0）举报 2021-06-06

我来回答

来自不同Kafka主题的事件数的聚合(总和)

1条答案

相关问题

热门标签

最新问答