Kafka Streams 应用程序 - 计数和总和聚合 [英] Kafka Streams App - count and sum aggregate
本文介绍了Kafka Streams 应用程序 - 计数和总和聚合的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我正在尝试从 KGroupedStream 创建一个 KTable 来存储每个键的值的总和.
I'm trying to create a KTable from a KGroupedStream to store the sum of the value for each key.
final StreamsBuilder builder = new StreamsBuilder();
final KTable<String, Long> sum = builder.stream("streams-plaintext-input")
.groupByKey()
.aggregate(new Initializer<Long>() {
@Override
public Long apply() {
return Long.MIN_VALUE;
}
}, new Aggregator<String, Long, Long>() {
@Override
public Long apply(final String key, final Long value,final Long aggregate) {
aggregate += value;
return aggregate;
}
}, Materialized.<String, Long, KeyValueStore<Byte, byte[]>>as("counts-store"));
但我收到错误:
The method aggregate(Initializer<VR>, Aggregator<? super Object,? super Object,VR>, Materialized<Object,VR,KeyValueStore<Bytes,byte[]>>) in the type KGroupedStream<Object,Object> is not applicable for the arguments (new Initializer<Long>(){}, new Aggregator<String,Long,Long>(){}, Materialized<String,Long,KeyValueStore<Byte,byte[]>>)
我见过的所有示例都将 Serde 作为第三个参数传递,但我已经尝试过并得到了一个非常相似的错误(我认为这可能来自旧版本,因为它与当前的签名不匹配)实施?):
All the examples I've seen pass in a Serde as the third argument but I have tried this and get a very similar error (I think this might be from an older version as it does not match with the signature of the current implementation?):
final StreamsBuilder builder = new StreamsBuilder();
final KTable<String, Long> sum = builder.stream("streams-plaintext-input")
.groupByKey()
.aggregate(new Initializer<Long>() {
@Override
public Long apply() {
return Long.MIN_VALUE;
}
}, new Aggregator<String, Long, Long>() {
@Override
public Long apply(final String key, final Long value,final Long aggregate) {
aggregate += value;
return aggregate;
}
}, Serdes.Long());
错误:
The method aggregate(Initializer<VR>, Aggregator<? super Object,? super Object,VR>, Materialized<Object,VR,KeyValueStore<Bytes,byte[]>>) in the type KGroupedStream<Object,Object> is not applicable for the arguments (new Initializer<Long>(){}, new Aggregator<String,Long,Long>(){}, Serde<Long>)
我做错了什么?
使用 Kafka 版本:2.1.0
Using Kafka version: 2.1.0
推荐答案
您的代码中存在一些问题:
There are few issues in your code:
- 对于
Materialized.as
而不是java.lang.Byte
你应该传递org.apache.kafka.common.utils.Bytes
立> - 你不应该修改
final
变量:aggregate += value;
- 您必须向
StreamsBuilder::stream
调用添加键和值的类型 (builder.
)stream("streams-plaintext-input")
- For
Materialized.as
insteadjava.lang.Byte
you should passorg.apache.kafka.common.utils.Bytes
- You shouldn't modify
final
variable:aggregate += value;
- You have to add types of key and value to
StreamsBuilder::stream
call (builder.<String, Long>stream("streams-plaintext-input")
)
修改后应该大致如下:
KTable<String, Long> sum = builder.<String, Long>stream("streams-plaintext-input")
.groupByKey()
.aggregate(new Initializer<Long>() {
@Override
public Long apply() {
return Long.MIN_VALUE;
}
}, new Aggregator<String, Long, Long>() {
@Override
public Long apply(final String key, final Long value,final Long aggregate) {
return aggregate + value;
}
}, Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"));
这篇关于Kafka Streams 应用程序 - 计数和总和聚合的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文