我对我的卡夫卡事件进行了分组:
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
但不是基于时间,而是基于事件的数量。
原因是事件将是批量转储,时间窗口聚合不适合,因为所有事件可能会在同一几秒钟内出现。
Kafka Streams 不支持开箱即用的基于计数的窗口,因为这些窗口是不确定的,并且很难处理无序数据。
不过,您可以使用处理器 API 为您的用例构建自定义运算符,而不是使用 DSL。
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)