scala - apache flink 的 union 类型混淆？

Question

我尝试合并多个 flink 数据集。它们包含在 Seq 中。以下是产生问题的代码

case class clickZap ( date: LocalDateTime, stbId:String, channelId :Int , nozap:Boolean)
val afterLastz: DataSet[clickZap]= ... 

val ma_range: IndexedSeq[DataSet[(Int, Option[(java.time.LocalDateTime, String, Int, Boolean)])]]  = for (i  <- Range (0,min_n))
      yield afterLastz.reduceGroup(it =>(i, maxBeforezTCZ(it,at plusMinutes(i))))
//val ma_all =  ma_range.slice(1, min_n).foldLeft(ma_range.head)(_ union _)
val ma_all = ma_range.head union(ma_range.tail.head)

我得到的是一个

Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: Cannot union inputs of different types. Input1=scala.Tuple2(_1: Integer, _2: Option[scala.Tuple4(_1: GenericType [java.time.LocalDateTime], _2: String, _3: Integer, _4: Boolean)]), input2=scala.Tuple2(_1: Integer, _2: Option[scala.Tuple4(_1: GenericType[java.time.LocalDateTime], _2: String, _3: Integer, _4: Boolean)])

我错过了什么？类型没有不同，不是吗？联合运算符应该很便宜，因此规避这个问题似乎没有吸引力。 我提供了前两行代码作为参数，表明数据集中的数据类型是相同的。 我使用的是flink版本0.9.0和0.9.1

Answer 1

问题是 Flink 自己的打字系统中的一个错误。表示 Scala Option 的 OptionTypeInfo 未定义正确的 equals 方法。因此，未检测到两个 OptionTypeInfos 相等。

我创建了一个JIRA issue并开通了Pull Request解决问题。拉取请求应在两天内合并。如果您随后使用最新的 0.10-SNAPSHOT 版本，那么您的问题应该得到解决。