除了DataStream 算子操作产生的主流之外,您还可以生成任意数量的附加旁路输出结果流。结果流中的数据类型不必与主流中的数据类型匹配,并且不同旁路输出的类型也可以不同。当您希望拆分通常必须复制流的数据流,然后从每个流中过滤掉您不希望拥有的数据时,此 算子操作非常有用。
使用旁路输出时,首先需要定义一个OutputTag用于标识旁路输出流的方法:
// this needs to be an anonymous inner class, so that we can analyze the typeOutputTag<String> outputTag = new OutputTag<String>("side-output") {};
val outputTag = OutputTag[String](docs_1.7-SNAPSHOT_"side-output")
注意如何OutputTag根据旁路输出流包含的数据元类型键入。
可以通过以下函数将数据发送到旁路输出:
- ProcessFunction
- CoProcessFunction
- ProcessWindowFunction
- ProcessAllWindowFunction
您可以使用Context在上述函数中向用户公开的参数将数据发送到由a标识的旁路输出OutputTag。以下是从以下位置发出旁路输出数据的示例ProcessFunction:
DataStream<Integer> input = ...;final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};SingleOutputStreamOperator<Integer> mainDataStream = input.process(new ProcessFunction<Integer, Integer>() {@Overridepublic void processElement(Integer value,Context ctx,Collector<Integer> out) throws Exception {// emit data to regular outputout.collect(value);// emit data to side outputctx.output(outputTag, "sideout-" + String.valueOf(value));}});
val input: DataStream[Int] = ...val outputTag = OutputTag[String](docs_1.7-SNAPSHOT_"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 outputout.collect(value)// emit data to side outputctx.output(outputTag, "sideout-" + String.valueOf(value))}})
用于检索您getSideOutput(OutputTag) 在DataStream 算子操作结果上使用的旁路输出流。这将为您提供一个DataStream输入到旁路输出流的结果:
final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};SingleOutputStreamOperator<Integer> mainDataStream = ...;DataStream<String> sideOutputStream = mainDataStream.getSideOutput(outputTag);
val outputTag = OutputTag[String](docs_1.7-SNAPSHOT_"side-output")val mainDataStream = ...val sideOutputStream: DataStream[String] = mainDataStream.getSideOutput(outputTag)
