Flink，为什么 CoMap 会返回“DataStream with Product with Serializable”而不仅仅是 DataStream？

Question

我需要了解为什么eventStream.connect(otherStream).map(_ => Right(2), _ => Left("2")) 不会生成DataStream[Either[String, Int]] 而是DataStream[Either[String, Int]] with Product with Serializable。 我正在使用一些接受DataStream[T] 的API，如果我将它们传递给DataStream[T] with Product with Serializable，则会出现编译时错误。有人可以解释一下，或许能给我一些提示吗？

我给你举个例子：

class FlinkFoo {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // Silly int source
    val eventStream: DataStream[Int] = env.addSource((sc: SourceContext[Int]) => {
      while (true) sc.collect(1)
    })

    // Silly String source
    val otherStream: DataStream[String] = env.addSource((sc: SourceContext[String]) => {
      while (true) sc.collect("1")
    })

    // I need to connect two stream and then flatten them
    val connectedStream2: DataStream[Either[String, Int] with Product with Serializable] = eventStream.connect(otherStream).map(_ => Right(2), _ => Left("2"))

    /* Compile time error !!!!
     * found   : org.apache.flink.streaming.api.scala.DataStream[Either[String,Int] with Product with Serializable]
     * [error]  required: org.apache.flink.streaming.api.scala.DataStream[Either[?,?]]
     * [error] Note: Either[String,Int] with Product with Serializable <: Either[?,?], but class DataStream is invariant in type T.
     * [error] You may wish to define T as +T instead. (SLS 4.5)
     * [error]     fooMethod(connectedStream2)
     * [error]               ^
     **/
    fooMethod(connectedStream2)
  }

  def fooMethod[T, P](dataStream: DataStream[Either[T, P]]): Unit = {
    // do something
  }
}

Answer 1

您可以尝试将 Flink scala 隐式序列化程序和 TypeInformation 添加到您的范围中，如下所示

import org.apache.flink.streaming.api.scala._

TypeUtilsobject被上面导入的包对象调用；它们为Either 提供序列化程序和所需的类型信息，与许多其他实体一样。

在 Flink 泛型类型解析之后，您需要这些转换来解析 Either 类型，并且您可能会显式地将返回类型添加到您的分配中以实现该转换。

val yourEitherStream: DataStream[Either[String, Int]] =
  eventStream
    .connect(otherStream)
    .map(_ => Right(2), _ => Left("2"))

with Product with Serializable mix-in 是 Scala 2.11 issue 的垃圾，在 2.12 中解决（但不能与 Flink right now 一起使用）。