Apache Flink 官方文檔--流(DataStream API)-旁路輸出

旁路輸出(side output)

??除了來自數據流算子的主流結果輸出之外，可以產生任意數量的流旁路輸出結果。旁路輸出結果數據類型與主流結果的數據類型以及其他旁路輸出結果數據類型可以是完全不同的。當你需要分割數據流時，這個算子非常有用。通常需要復制流，然后從每個數據流中過濾掉不需要的數據。
??當使用旁路輸出時，首先需要定義一個OutputTag來標識一個旁路輸出流。
Java

// this needs to be an anonymous inner class, so that we can analyze the type
OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

Scala

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

??注意OutputTag是如何根據旁路輸出流包含的元素類型typed的。
??可以通過以下函數發射數據到旁路輸出。

• ProcessFunction
• CoProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

??可以使用Context參數（在上述函數中向用戶暴露）將數據發送到OutputTag標識的旁路輸出。以下是從ProcessFunction發出旁路輸出數據的示例：
Java:

DataStream<Integer> input = ...;

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = input
  .process(new ProcessFunction<Integer, Integer>() {

      @Override
      public void processElement(
          Integer value,
          Context ctx,
          Collector<Integer> out) throws Exception {
        // emit data to regular output
        out.collect(value);

        // emit data to side output
        ctx.output(outputTag, "sideout-" + String.valueOf(value));
      }
    });

Scala:

val input: DataStream[Int] = ...
val outputTag = 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 = {
      // emit data to regular output
      out.collect(value)

      // emit data to side output
      ctx.output(outputTag, "sideout-" + String.valueOf(value))
    }
  })

  要讀取旁路輸出流，在數據流運算后使用getSideOutput(OutputTag)。此時將會獲得鍵入旁路輸出流的結果。
Java:

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = ...;

DataStream<String> sideOutputStream = mainDataStream.getSideOutput(outputTag);

Scala:

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

val mainDataStream = ...

val sideOutputStream: DataStream[String] = mainDataStream.getSideOutput(outputTag)