事件计数的窗口聚合

Question

我已经将我的kafka事件分组：

private static void createImportStream(final StreamsBuilder builder, final Collection<String> topics) {
final KStream<byte[], GraphEvent> stream = builder.stream(topics, Consumed.with(Serdes.ByteArray(), new UserEventThriftSerde()));
stream.filter((key, request) -> {
return Objects.nonNull(request);
}).groupBy(
(key, value) -> Integer.valueOf(value.getSourceType()),
Grouped.with(Serdes.Integer(), new UserEventThriftSerde()))
.aggregate(ArrayList::new, (key, value, aggregatedValue) -> {
aggregatedValue.add(value);
return aggregatedValue;
},
Materialized.with(Serdes.Integer(), new ArrayListSerde<UserEvent>(new UserEventThriftSerde()))
).toStream();
}

如何添加window，但不是基于时间，而是基于事件数量。原因是这些事件将是大容量转储，时间窗口聚合不适合，因为所有事件都可能在同一秒内出现。

Answer 1

Kafka Streams不支持开箱即用的基于计数的窗口，因为这些窗口是不确定的，很难处理无序的数据。

您可以使用Processor API为您的用例构建自定义运算符，而不是使用DSL。