Apache Flink公式ドキュメント--ストリーム(DataStream API)-バイパス出力

バイパス出力
  データストリーム演算子からの主流結果出力に加えて、任意の数のストリームバイパス出力結果を生成することができる.バイパス出力結果データ型は、主流結果のデータ型および他のバイパス出力結果データ型とは全く異なる場合がある.この演算子は、データストリームを分割する必要がある場合に便利です.通常、ストリームをコピーし、不要なデータを各データストリームからフィルタリングする必要があります. バイパス出力を使用する場合、まず、バイパス出力ストリームを識別するためにOutputTagを定義する必要がある.Java

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

Scala

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

OutputTagは、バイパス出力ストリームに含まれる要素タイプtypedに従ってどのように出力されるかに注意してください.  は、以下の関数によってバイパス出力にデータを送信することができる.

ProcessFunction

CoProcessFunction

ProcessWindowFunction

ProcessAllWindowFunction

  Contextパラメータを使用して、OutputTag識別子のバイパス出力にデータを送信することができる.次の例は、ProcessFunctionからバイパス出力データを発行する例です:Java:

DataStream input = ...;

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

SingleOutputStreamOperator mainDataStream = input
  .process(new ProcessFunction() {

      @Override
      public void processElement(
          Integer value,
          Context ctx,
          Collector 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 outputTag = new OutputTag("side-output"){};

SingleOutputStreamOperator mainDataStream = ...;

DataStream sideOutputStream = mainDataStream.getSideOutput(outputTag);

Scala:

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

val mainDataStream = ...

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