重载方法值与替代方法 [英] overloaded method value aggregate with alternatives
本文介绍了重载方法值与替代方法的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我具有以下无法编译的功能:
I have following function, that does not compile:
private def save(pea: KStream[String, String]): Unit = {
pea
.groupByKey()
.aggregate(() => """{folder: ""}""",
(_: String, _: String, value: String) => value,
EventStoreTopology.Store)
}
错误消息是:
[error] [VR](x$1: org.apache.kafka.streams.kstream.Initializer[VR], x$2: org.apache.kafka.streams.kstream.Aggregator[_ >: String, _ >: String, VR], x$3: org.apache.kafka.streams.processor.StateStoreSupplier[org.apache.kafka.streams.state.KeyValueStore[_, _]])org.apache.kafka.streams.kstream.KTable[String,VR] <and>
[error] [VR](x$1: org.apache.kafka.streams.kstream.Initializer[VR], x$2: org.apache.kafka.streams.kstream.Aggregator[_ >: String, _ >: String, VR], x$3: org.apache.kafka.common.serialization.Serde[VR])org.apache.kafka.streams.kstream.KTable[String,VR] <and>
[error] [VR](x$1: org.apache.kafka.streams.kstream.Initializer[VR], x$2: org.apache.kafka.streams.kstream.Aggregator[_ >: String, _ >: String, VR], x$3: org.apache.kafka.streams.kstream.Materialized[String,VR,org.apache.kafka.streams.state.KeyValueStore[org.apache.kafka.common.utils.Bytes,Array[Byte]]])org.apache.kafka.streams.kstream.KTable[String,VR]
[error] cannot be applied to (() => String, (String, String, String) => String, io.khinkali.eventstore.EventStoreTopology.Persistent)
[error] .aggregate(() => """{folder: ""}""",
[error] ^
[error] one error found
[error] (eventstore/compile:compileIncremental) Compilation failed
aggregate的签名是:
the signature of aggregate is:
<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
final Aggregator<? super K, ? super V, VR> aggregator,
final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
并且 EventStoreTopology.Store
定义为:
object EventStoreTopology {
type Persistent = Materialized[String, String, KeyValueStore[Bytes, Array[Byte]]]
val StoreName: String = "EventStore"
val Store: Persistent = Materialized.as(StoreName)
}
我在做什么错了?
推荐答案
编译器需要一些帮助来推断 aggregator
参数的正确类型.
The compiler need some help to infer the correct type for the aggregator
parameter.
要使其编译,您可以尝试:
To make it compile you can try:
val store: Materialized[String, String, KeyValueStore[Bytes, Array[Byte]]] = ???
private def save(pea: KStream[String, String]): Unit = {
val aggregator: Aggregator[String, String, String] = (_, _, value: String) => value
pea
.groupByKey()
.aggregate(() => """{folder: ""}""",
aggregator,
store)
}
这篇关于重载方法值与替代方法的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文