重载方法值与替代项聚合

aor9mmx1 于 2021-06-07 发布在 Kafka

关注(0)|答案(1)|浏览(398)

我有以下函数，它不编译：

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的签名为：

<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)

}

我做错什么了？

scala apache-kafka

来源：https://stackoverflow.com/questions/48145527/overloaded-method-value-aggregate-with-alternatives

1条答案

按热度按时间

eufgjt7s1#

编译器需要一些帮助来推断 aggregator 参数。
要使其编译，您可以尝试：

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)
}

赞(0）回复(0）举报 2021-06-07

我来回答

重载方法值与替代项聚合

1条答案

相关问题

热门标签

最新问答