如何每小时汇总数据？

Question

每当用户喜欢我们网站上的某些内容时，我们都会收集事件，我们计划做的是每小时提交内容的汇总收藏夹并更新数据库中的收藏夹总数。

我们正在评估 Kafka Streams。遵循字数统计示例。我们的拓扑很简单，生成到一个主题 A 并读取聚合数据并将其提交到另一个主题 B。然后每小时消耗来自主题 B 的事件并提交到数据库中。

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
   public StreamsConfig kStreamsConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "favorite-streams");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokerAddress);
    return new StreamsConfig(props);
}

@Bean
public KStream<String, String> kStream(StreamsBuilder kStreamBuilder) {
    StreamsBuilder builder = streamBuilder();
    KStream<String, String> source = builder.stream(topic);
    source.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split("\\W+")))
            .groupBy((key, value) -> value)
            .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>> as("counts-store")).toStream()
            .to(topic + "-grouped", Produced.with(Serdes.String(), Serdes.Long()));

    Topology topology = builder.build();
    KafkaStreams streams = new KafkaStreams(topology, kStreamsConfigs());
    streams.start();
    return source;
}

@Bean
public StreamsBuilder streamBuilder() {
    return new StreamsBuilder();
}

然而，当我使用这个主题 B 时，它从一开始就为我提供了聚合数据。我的问题是，我们是否可以提供一些规定，我可以使用前几个小时分组的数据，然后提交到数据库，然后 Kakfa 忘记前几个小时的数据，并每小时提供新数据而不是累积总和。设计拓扑是正确的还是我们可以做得更好？

Answer 1

如果要每小时获得一个聚合结果，可以使用窗口大小为 1 小时的窗口聚合。

stream.groupBy(...)
      .windowedBy(TimeWindow.of(1 *3600 * 1000))
      .count(...)

查看文档以获取更多详细信息：https : //docs.confluent.io/current/streams/developer-guide/dsl-api.html#windowing

输出类型Windowed<String>用于键（不是String）。您需要提供自定义Window<String>Serde，或转换密钥类型。请参阅 SessionWindowsExample。