如何在scala中使用flink折叠功能

Question

这是将 Flink fold 与 scala 匿名函数一起使用的无效尝试：

val myFoldFunction = (x: Double, t:(Double,String,String)) => x + t._1
env.readFileStream(...).
...
.groupBy(1)
.fold(0.0, myFoldFunction : Function2[Double, (Double,String,String), Double])

它编译得很好，但在执行时，我得到一个 "type erasure issue"（见下文）。在 Java 中这样做很好，但当然更冗长。我喜欢简洁明了的 lambda。我怎样才能在 Scala 中做到这一点？

Caused by: org.apache.flink.api.common.functions.InvalidTypesException:
Type of TypeVariable 'R' in 'public org.apache.flink.streaming.api.scala.DataStream org.apache.flink.streaming.api.scala.DataStream.fold(java.lang.Object,scala.Function2,org.apache.flink.api.common.typeinfo.TypeInformation,scala.reflect.ClassTag)' could not be determined. 
This is most likely a type erasure problem. 
The type extraction currently supports types with generic variables only in cases where all variables in the return type can be deduced from the input type(s).

Answer 1

您遇到的问题是Flink[1]中的一个bug。问题源于 Flink 的 TypeExtractor 以及 Scala DataStream API 在 Java 实现之上实现的方式。 TypeExtractor 无法为 Scala 类型生成 TypeInformation，因此 returns 无法生成 MissingTypeInformation。这个缺失的类型信息是在创建 StreamFold 运算符后手动设置的。但是，StreamFold 运算符以不接受 MissingTypeInformation 的方式实现，因此在设置正确的类型信息之前失败。

我已经打开了一个拉取请求 [2] 来解决这个问题。它应该在接下来的两天内合并。通过使用最新的 0.10 快照版本，您的问题应该得到解决。

如何在scala中使用flink折叠功能

How to use flink fold function in scala

scala

apache-flink