时间戳

首页 > 解决方案 > Apache Kafka 分组两次

apache-kafka - Apache Kafka 分组两次

问题描述

我正在编写一个应用程序，我试图计算每小时访问一个页面的用户数量。我正在尝试过滤到特定事件，按 userId 和事件小时时间分组，然后按小时分组以获取用户数。但是在尝试关闭流时，对 KTable 进行分组会导致过多的 CPU 消耗和锁定。有一个更好的方法吗？

    events
   .groupBy(...)
   .aggregate(...)
   .groupBy(...);
   .count();

标签： apache-kafkaapache-kafka-streams

解决方案

鉴于上述问题的答案“我只想在一个小时的时间窗口内知道执行特定操作的用户数量”，我会建议以下内容。

假设您有这样的记录：

class ActionRecord {
  String actionType;
  String user;
}

您可以定义一个聚合类，如下所示：

class ActionRecordAggregate {
  private Set<String> users = new HashSet<>();

  public void add(ActionRecord rec) {
    users.add(rec.getUser());
  }

  public int count() {
    return users.size();
  }

}

然后您的流媒体应用程序可以：

  • 接受事件
  • 根据事件类型重新设置密钥（.map()）
  • 按事件类型分组 ( .groupByKey())
  • 按时间窗口（选择 1 分钟但 YMMV）
  • 将它们聚合成ActionRecordAggregate
  • 将它们具体化为 StateStore

所以这看起来像：

stream()
.map((key, val) -> KeyValue.pair(val.actionType, val)) 
.groupByKey() 
.windowedBy(TimeWindows.of(60*1000)) 
.aggregate(
  ActionRecordAggregate::new, 
  (key, value, agg) -> agg.add(value),
  Materialized
      .<String, ActionRecordAggregate, WindowStore<Bytes, byte[]>>as("actionTypeLookup")
      .withValueSerde(getSerdeForActionRecordAggregate())
);

然后，要取回事件，您可以查询您的状态存储：

ReadOnlyWindowStore<String, ActionRecordAggregate> store = 
  streams.store("actionTypeLookup", QueryableStoreTypes.windowStore());

WindowStoreIterator<ActionRecordAggregate> wIt = 
  store.fetch("actionTypeToGet", startTimestamp, endTimestamp);

int totalCount = 0;
while(wIt.hasNext()) {
  totalCount += wIt.next().count();
}

// totalCount is the number of distinct users in your 
// time interval that raised action type "actionTypeToGet"

希望这可以帮助！

推荐阅读