gpt4 book ai didi

scala - flatMap 函数中的 Apache Flink Streaming 类型不匹配

转载 作者:行者123 更新时间:2023-12-04 17:53:16 34 4
gpt4 key购买 nike

尝试在 scala 2.10.4 中使用 0.10.0 flink 版本的流 api。在尝试编译第一个版本时:

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.windowing.time._

object Main {
def main(args: Array[String]) {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val text = env.socketTextStream("localhost", 9999)

val words : DataStream[String] = text.flatMap[String](
new Function[String,TraversableOnce[String]] {
def apply(line:String):TraversableOnce[String] = line.split(" ")
})

env.execute("Window Stream wordcount")
}
}

我收到编译时错误:

[error]  found   : String => TraversableOnce[String]
[error] required: org.apache.flink.api.common.functions.FlatMapFunction[String,String]
[error] new Function[String,TraversableOnce[String]] { def apply(line:String):TraversableOnce[String] = line.split(" ")})
[error] ^

在我包含在项目中的 DataStream.class 的反编译版本中,有一些函数可以接受这种类型(最后一个):

public <R> DataStream<R> flatMap(FlatMapFunction<T, R> flatMapper, TypeInformation<R> evidence$12, ClassTag<R> evidence$13) {
if (flatMapper == null) {
throw new NullPointerException("FlatMap function must not be null.");
}
TypeInformation outType = (TypeInformation)Predef..MODULE$.implicitly(evidence$12);
return package..MODULE$.javaToScalaStream((org.apache.flink.streaming.api.datastream.DataStream)this.javaStream.flatMap(flatMapper).returns(outType));
}

public <R> DataStream<R> flatMap(Function2<T, Collector<R>, BoxedUnit> fun, TypeInformation<R> evidence$14, ClassTag<R> evidence$15) {
if (fun == null) {
throw new NullPointerException("FlatMap function must not be null.");
}
Function2<T, Collector<R>, BoxedUnit> cleanFun = this.clean((F)fun);
.anon flatMapper = new /* Unavailable Anonymous Inner Class!! */;
return this.flatMap((FlatMapFunction<T, R>)flatMapper, evidence$14, evidence$15);
}

public <R> DataStream<R> flatMap(Function1<T, TraversableOnce<R>> fun, TypeInformation<R> evidence$16, ClassTag<R> evidence$17) {
if (fun == null) {
throw new NullPointerException("FlatMap function must not be null.");
}
Function1<T, TraversableOnce<R>> cleanFun = this.clean((F)fun);
.anon flatMapper = new /* Unavailable Anonymous Inner Class!! */;
return this.flatMap((FlatMapFunction<T, R>)flatMapper, evidence$16, evidence$17);
}

这里可能有什么问题?如果您能提供一些见解,我将不胜感激。提前谢谢你。

最佳答案

问题是你正在导入Flink的Java StreamExecutionEnvironment:org.apache.flink.streaming.api.environment.StreamExecutionEnvironment

您必须像这样使用 StreamExecutionEnvironment 的 Scala 变体:import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment。有了这个改变,一切都在成功构建!

原答案:问题是您将 Function 传递给 flatMap() 方法。但是 flatMap() 需要一个 FlatMapFunction

 val words : DataStream[String] = text.flatMap[String](
new FlatMapFunction[String,String] {
override def flatMap(t: String, collector: Collector[String]): Unit = t.split(" ")
})

关于scala - flatMap 函数中的 Apache Flink Streaming 类型不匹配,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/33720947/

34 4 0