-1

例如每个客户的用电量。消费数据并非一直在流式传输，而是按一定数量的客户在过去 12 小时内分批插入。插入后，我们需要汇总每个客户的每小时消耗量，如果某些客户的前一小时消耗量不存在，以找到“最接近日期”的消耗量。

4

1 回答 1

1

您是否查看了 Kafka Streams ( https://kafka.apache.org/documentation/streams/ )？

它允许您将主题作为数据流读取并在时间窗口上聚合：

StreamsBuilder builder = new StreamsBuilder();
builder.stream("topic-name")
  .groupByKey() // assuming the key is a customer-ID
  .windowedBy(TimeWindows.of(Duration.ofHours(1)))
  .aggregate(...); // insert business logic here
于 2018-11-24T21:56:19.693 回答