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

旁路輸出(side output)

??除了來自數(shù)據(jù)流算子的主流結(jié)果輸出之外，可以產(chǎn)生任意數(shù)量的流旁路輸出結(jié)果。旁路輸出結(jié)果數(shù)據(jù)類型與主流結(jié)果的數(shù)據(jù)類型以及其他旁路輸出結(jié)果數(shù)據(jù)類型可以是完全不同的。當(dāng)你需要分割數(shù)據(jù)流時，這個算子非常有用。通常需要復(fù)制流，然后從每個數(shù)據(jù)流中過濾掉不需要的數(shù)據(jù)。
??當(dāng)使用旁路輸出時，首先需要定義一個OutputTag來標(biāo)識一個旁路輸出流。
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是如何根據(jù)旁路輸出流包含的元素類型typed的。
??可以通過以下函數(shù)發(fā)射數(shù)據(jù)到旁路輸出。

• ProcessFunction
• CoProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

??可以使用Context參數(shù)（在上述函數(shù)中向用戶暴露）將數(shù)據(jù)發(fā)送到OutputTag標(biāo)識的旁路輸出。以下是從ProcessFunction發(fā)出旁路輸出數(shù)據(jù)的示例：
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))
    }
  })

  要讀取旁路輸出流，在數(shù)據(jù)流運算后使用getSideOutput(OutputTag)。此時將會獲得鍵入旁路輸出流的結(jié)果。
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)