java - 最后 X 分钟的 Kafka 实时平均值

Question

我有一个关于单个代理的主题，该代理不断传入有关用户点击的数据。我希望能够近乎实时地(例如 1 秒)计算每个用户在最后 X 分钟内的平均点击次数。

我尝试用 Kafka 流来做，但问题是缩略图窗口无法近乎实时地计算，并且每秒更新最后 X 分钟内所有值的平均值。跳跃窗口可能适合，但对于跳跃 = 1 秒和大小为 5 分钟的窗口，它将创建 300 个窗口，我想从性能方面来看它太多了。

有没有办法在没有像 Spark 这样的第 3 方流媒体引擎的情况下使用“普通”Kafka API 来做到这一点？ (Kafka 流是可选的)。

非常感谢!

Answer 1

正如评论者所说 - 使用 Kafka Streams , 或 KSQL . KSQL 在 Kafka Streams 之上运行，因此数据建模和窗口和聚合等概念是相同的。

在 KSQL 中:

ksql> CREATE TABLE USER_CLICKS_PER_MINUTE AS \
SELECT USER_ID, COUNT(*) AS CLICK_COUNT, \
COUNT(*)/5 AS CLICKS_PER_MINUTE \
FROM RATINGS WINDOW HOPPING (SIZE 5 MINUTES, ADVANCE BY 1 SECOND) \
GROUP BY USER_ID;

 Message
---------------------------
 Table created and running
---------------------------

查询有状态聚合:

ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss'), USER_ID, \
CLICK_COUNT, CLICKS_PER_MINUTE \
FROM USER_CLICKS_PER_MINUTE \
WHERE USER_ID=18;
2018-05-09 11:44:33 | 18 | 7 | 1
2018-05-09 11:44:34 | 18 | 7 | 1
2018-05-09 11:44:35 | 18 | 7 | 1
2018-05-09 11:44:36 | 18 | 9 | 1
2018-05-09 11:44:37 | 18 | 9 | 1
2018-05-09 11:44:38 | 18 | 10 | 2
2018-05-09 11:44:39 | 18 | 10 | 2
2018-05-09 11:44:40 | 18 | 10 | 2
2018-05-09 11:44:41 | 18 | 12 | 2
2018-05-09 11:44:42 | 18 | 12 | 2
2018-05-09 11:44:43 | 18 | 12 | 2
2018-05-09 11:44:44 | 18 | 12 | 2
2018-05-09 11:44:45 | 18 | 12 | 2
2018-05-09 11:44:46 | 18 | 12 | 2
2018-05-09 11:44:47 | 18 | 12 | 2
2018-05-09 11:44:48 | 18 | 12 | 2
2018-05-09 11:44:49 | 18 | 12 | 2
2018-05-09 11:44:50 | 18 | 12 | 2
2018-05-09 11:44:51 | 18 | 13 | 2
2018-05-09 11:44:52 | 18 | 13 | 2
2018-05-09 11:44:53 | 18 | 13 | 2
2018-05-09 11:44:54 | 18 | 13 | 2
2018-05-09 11:44:55 | 18 | 13 | 2
2018-05-09 11:44:56 | 18 | 13 | 2
2018-05-09 11:44:57 | 18 | 13 | 2
2018-05-09 11:44:58 | 18 | 13 | 2
2018-05-09 11:44:59 | 18 | 13 | 2
2018-05-09 11:45:00 | 18 | 13 | 2
2018-05-09 11:45:01 | 18 | 13 | 2

请记住，Kafka Streams 和 KSQL(基于 Kafka Streams 构建)将在新事件到达时重新发出给定时间窗口的聚合。根据您的要求，每 1 秒前进一次的实际跳跃窗口可能不是您想要的。一个实时更新的简单滚动窗口仍然会为您提供有状态聚合的实时结果。

更多关于 KSQL 的信息:

• > https://www.confluent.io/product/ksql/
• > KSQL docs
• > KSQL syntax reference
• > KSQL Quickstart tutorial
• > KSQL video tutorials