我正在尝试将基于 Scala 的 Kafka Streams 应用程序从 0.10.0.0 升级到 0.10.2.1,我无法弄清楚如何让应用程序进行编译。
我在文件中找到的示例使用mapValue
但它不会更改值类型。我按照这个使用带有-Xexperimental
编译器标志的 Scala 2.11.
码
class MyStream() {
def startMyStream(): Unit = {
val kStreamBuilder = new KStreamBuilder
val kStream = kStreamBuilder.stream("myTopic")
kStream.mapValues(new ValueMapper[AnyRef, Double]() {
override def apply(value: Any) = 6.3
})
val kafkaStreams = new KafkaStreams(kStreamBuilder, new Properties)
kafkaStreams.start()
}
}
编译器错误
no type parameters for method mapValues: (x$1: org.apache.kafka.streams.kstream.ValueMapper[_, _ <: VR])org.apache.kafka.streams.kstream.KStream[Nothing,VR] exist so that it can be applied to arguments (org.apache.kafka.streams.kstream.ValueMapper[AnyRef,Double]{})
--- because ---
argument expression's type is not compatible with formal parameter type;
found : org.apache.kafka.streams.kstream.ValueMapper[AnyRef,Double]
required: org.apache.kafka.streams.kstream.ValueMapper[_, _ <: ?VR]
Note: AnyRef <: Any, but Java-defined trait ValueMapper is invariant in type V.
You may wish to investigate a wildcard type such as `_ <: Any`. (SLS 3.2.10)
kStream.mapValues(new ValueMapper[AnyRef, Double]() {
^
type mismatch;
found : org.apache.kafka.streams.kstream.ValueMapper[AnyRef,Double]{}
required: org.apache.kafka.streams.kstream.ValueMapper[_, _ <: VR]
kStream.mapValues(new ValueMapper[AnyRef, Double]() {
^
two errors found
表示为 java 类,编译很好。
public class MyStream {
public void startMyStream() {
KStreamBuilder kStreamBuilder = new KStreamBuilder();
KStream kStream = kStreamBuilder.stream("myTopic");
kStream.mapValues(new ValueMapper<Object, Double>() {
@Override
public Double apply(Object value) {
return 6.3;
}
});
KafkaStreams kafkaStreams = new KafkaStreams(kStreamBuilder, new Properties());
kafkaStreams.start();
}
}
如何编译 Scala 版本?
基于此答案的解决方案
三种工作方式,两种方式不工作。
class MyStream() {
def startMyStream(): Unit = {
val kStreamBuilder = new KStreamBuilder
// Explicit tying here is not required to compile and run.
val kStream: KStream[Array[Byte], String] = kStreamBuilder.stream("myTopic")
// Does not compile
kStream.mapValues(new ValueMapper[AnyRef, Double]() {
override def apply(value: AnyRef) = 6.3
})
// Does not compile
kStream.mapValues(_ => 6.3)
// Works
kStream.mapValues[Double](new ValueMapper[AnyRef, Double]() {
override def apply(value: AnyRef) = 6.3
})
// Works, requires compiler option -Xexperimental
kStream.mapValues[Double](_ => 6.3)
// Works, requires compiler option -Xexperimental
kStream.mapValues[Double](convert)
def convert(string: String): Double = 6.3
val kafkaStreams = new KafkaStreams(kStreamBuilder, new Properties)
kafkaStreams.start()
}
}
更新:无效的解决方案
尝试 1按建议(val kStream: KStream[Array[Byte], String] = kStreamBuilder.stream("myTopic")
)向 kStream 添加显式类型仍然无法编译并导致此错误。
no type parameters for method mapValues: (x$1: org.apache.kafka.streams.kstream.ValueMapper[_ >: String, _ <: VR])org.apache.kafka.streams.kstream.KStream[Array[Byte],VR] exist so that it can be applied to arguments (org.apache.kafka.streams.kstream.ValueMapper[AnyRef,Double]{})
--- because ---
argument expression's type is not compatible with formal parameter type;
found : org.apache.kafka.streams.kstream.ValueMapper[AnyRef,Double]
required: org.apache.kafka.streams.kstream.ValueMapper[_ >: String, _ <: ?VR]
Note: AnyRef <: Any, but Java-defined trait ValueMapper is invariant in type V.
You may wish to investigate a wildcard type such as `_ <: Any`. (SLS 3.2.10)
kStream.mapValues(new ValueMapper[AnyRef, Double]() {
^
type mismatch;
found : org.apache.kafka.streams.kstream.ValueMapper[AnyRef,Double]{}
required: org.apache.kafka.streams.kstream.ValueMapper[_ >: String, _ <: VR]
kStream.mapValues(new ValueMapper[AnyRef, Double]() {
尝试 2添加上述内容并使用 SAM 转换“以避免显式写出匿名类的实例化”(kStream.mapValues(_ => 6.3)
)导致此编译器错误。
no type parameters for method mapValues: (x$1: org.apache.kafka.streams.kstream.ValueMapper[_ >: String, _ <: VR])org.apache.kafka.streams.kstream.KStream[Array[Byte],VR] exist so that it can be applied to arguments (org.apache.kafka.streams.kstream.ValueMapper[String,Double] with Serializable)
--- because ---
argument expression's type is not compatible with formal parameter type;
found : org.apache.kafka.streams.kstream.ValueMapper[String,Double] with Serializable
required: org.apache.kafka.streams.kstream.ValueMapper[_ >: String, _ <: ?VR]
Note: String <: Any (and org.apache.kafka.streams.kstream.ValueMapper[String,Double] with Serializable <: org.apache.kafka.streams.kstream.ValueMapper[String,Double]), but Java-defined trait ValueMapper is invariant in type V.
You may wish to investigate a wildcard type such as `_ <: Any`. (SLS 3.2.10)
kStream.mapValues(_ => 6.3)
^
type mismatch;
found : org.apache.kafka.streams.kstream.ValueMapper[String,Double] with Serializable
required: org.apache.kafka.streams.kstream.ValueMapper[_ >: String, _ <: VR]
kStream.mapValues(_ => 6.3)
^
1 回答
这里有两个问题。首先,您不指定流的类型:
这实际上是您看到错误的原因 - 因为您没有指定类型,Scala 会将它们推断为某些默认值,这几乎总是不是您想要的。
其次,由于您已启用
-Xexperimental
,因此您可以依赖 SAM 转换来避免显式写出匿名类的实例化:**更新:**似乎由于某种原因,Scala 编译器无法正确推断匿名 function/SAM 实例的输出类型。我能够通过以下小调整成功编译代码: