Flink DataStream Side Output

Side Output 功能从 Flink 1.3.0 版本开始提供，FLINK-4460

除了从 DataStream 操作的结果中获取主数据流之外，还可以产生任意数量额外的旁路输出（side output）结果流。旁路输出的数据类型不需要与主数据流的类型一致，不同旁路输出的类型也可以不同。当你想要拆分数据流时（通常需要复制流），然后从每个流过滤出不需要的数据，此操作将非常有用。

当使用旁路输出时，首先需要定义一个 OutputTag 对象，用来标识一个旁路输出流:

val outputTag = OutputTag[String]("side-output")

可以通过以下函数将数据发送到旁路输出：

• ProcessFunction
• CoProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

可以使用 Context （在上述函数中会向用户提供使用）将数据发送到由 OutputTag 标识的旁路输出。下面的例子是在 ProcessFunction 中发出旁路输出数据：

val input: DataStream[Int] = ...

// 定义 OutputTag
val outputTag = new OutputTag[String]("side-output")

val mainDataStream = input.process(new ProcessFunction[Int, Int] {
    override def processElement(
        value: Int,
        ctx: ProcessFunction[Int, Int]#Context,
        out: Collector[Int]): Unit = {
      
      // 将数据发送到常规输出中
      out.collect(value)

      // 将数据发送到旁路输出中
      ctx.output(outputTag, "sideout-" + String.valueOf(value))
    }
  })

要获取旁路输出流，可以在 DataStream 上调用 getSideOutput(OutputTag) 来获取，如下面的例子：

val outputTag = OutputTag[String]("side-output")
val mainDataStream = ...
val sideOutputStream: DataStream[String] = mainDataStream.getSideOutput(outputTag)

Reference：
https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/stream/side_output.html