旁路输出

除了由 DataStream 操作产生的主要流之外,你还可以产生任意数量的旁路输出结果流。结果流中的数据类型不必与主要流中的数据类型相匹配,并且不同旁路输出的类型也可以不同。当你需要拆分数据流时,通常必须复制该数据流,然后从每个流中过滤掉不需要的数据,这个操作十分有用。

使用旁路输出时,首先需要定义用于标识旁路输出流的 OutputTag

Java

  1. // 这需要是一个匿名的内部类,以便我们分析类型
  2. OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

Scala

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

注意 OutputTag 是如何根据旁路输出流所包含的元素类型进行类型化的。

可以通过以下方法将数据发送到旁路输出:

你可以使用在上述方法中向用户暴露的 Context 参数,将数据发送到由 OutputTag 标识的旁路输出。这是从 ProcessFunction 发送数据到旁路输出的示例:

Java

  1. DataStream<Integer> input = ...;
  2. final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};
  3. SingleOutputStreamOperator<Integer> mainDataStream = input
  4. .process(new ProcessFunction<Integer, Integer>() {
  5. @Override
  6. public void processElement(
  7. Integer value,
  8. Context ctx,
  9. Collector<Integer> out) throws Exception {
  10. // 发送数据到主要的输出
  11. out.collect(value);
  12. // 发送数据到旁路输出
  13. ctx.output(outputTag, "sideout-" + String.valueOf(value));
  14. }
  15. });

Scala

  1. val input: DataStream[Int] = ...
  2. val outputTag = OutputTag[String]("side-output")
  3. val mainDataStream = input
  4. .process(new ProcessFunction[Int, Int] {
  5. override def processElement(
  6. value: Int,
  7. ctx: ProcessFunction[Int, Int]#Context,
  8. out: Collector[Int]): Unit = {
  9. // 发送数据到主要的输出
  10. out.collect(value)
  11. // 发送数据到旁路输出
  12. ctx.output(outputTag, "sideout-" + String.valueOf(value))
  13. }
  14. })

你可以在 DataStream 运算结果上使用 getSideOutput(OutputTag) 方法获取旁路输出流。这将产生一个与旁路输出流结果类型一致的 DataStream

Java

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

Scala

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