旁路输出

除了DataStream 算子操作产生的主流之外，您还可以生成任意数量的附加旁路输出结果流。结果流中的数据类型不必与主流中的数据类型匹配，并且不同旁路输出的类型也可以不同。当您希望拆分通常必须复制流的数据流，然后从每个流中过滤掉您不希望拥有的数据时，此 算子操作非常有用。

使用旁路输出时，首先需要定义一个OutputTag用于标识旁路输出流的方法：

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

注意如何OutputTag根据旁路输出流包含的数据元类型键入。

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

  • ProcessFunction
  • CoProcessFunction
  • ProcessWindowFunction

  • ProcessAllWindowFunction您可以使用Context在上述函数中向用户公开的参数将数据发送到由a标识的旁路输出OutputTag。以下是从以下位置发出旁路输出数据的示例ProcessFunction：

  • Java

  • Scala
  1. DataStream<Integer> input = ...;
  3. final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};
  5. SingleOutputStreamOperator<Integer> mainDataStream = input
  6.   .process(new ProcessFunction<Integer, Integer>() {
  8.       @Override
  9.       public void processElement(
  10.           Integer value,
  11.           Context ctx,
  12.           Collector<Integer> out) throws Exception {
  13.         // emit data to regular output
  14.         out.collect(value);
  16.         // emit data to side output
  17.         ctx.output(outputTag, "sideout-" + String.valueOf(value));
  18.       }
  19.     });
  1. val input: DataStream[Int] = ...
  2. val outputTag = OutputTag[String]("side-output")
  4. val mainDataStream = input
  5.   .process(new ProcessFunction[Int, Int] {
  6.     override def processElement(
  7.         value: Int,
  8.         ctx: ProcessFunction[Int, Int]#Context,
  9.         out: Collector[Int]): Unit = {
  10.       // emit data to regular output
  11.       out.collect(value)
  13.       // emit data to side output
  14.       ctx.output(outputTag, "sideout-" + String.valueOf(value))
  15.     }
  16.   })

用于检索您getSideOutput(OutputTag)在DataStream 算子操作结果上使用的旁路输出流。这将为您提供一个DataStream输入到旁路输出流的结果：

  1. final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};
  3. SingleOutputStreamOperator<Integer> mainDataStream = ...;
  5. DataStream<String> sideOutputStream = mainDataStream.getSideOutput(outputTag);
  1. val outputTag = OutputTag[String]("side-output")
  3. val mainDataStream = ...
  5. val sideOutputStream: DataStream[String] = mainDataStream.getSideOutput(outputTag)