使用旁路输出(side output)来拆分和复制流
我们在处理数据的时候，有时候想对不同情况的数据进行不同的处理，那么就需要把流进行拆分或者复制。
如果是使用filter来进行拆分，也能满足我们的需求，但每次筛选都要保留整个流，然后遍历整个流，显然很浪费性能，假如能够在一个流了多次输出就好了，flink的旁路输出则提供了这样的功能。
如何使用旁路输出呢
一、要定义OutputTag，这个就类似一个容器，需要拆分成几个流，就定义几个
//例如定义一个tag,来收集某个值小于200的数据
private static final OutputTag<Order> outputTag1 = new OutputTag<>("side-output<200", TypeInformation.of(Order.class));
private static final OutputTag<Order> outputTag2 = new OutputTag<>("side-output-200~500", TypeInformation.of(Order.class));
private static final OutputTag<String> outputTag3 = new OutputTag<>("side-output>500", TypeInformation.of(String.class));
二、使用ProcessFunction来进行拆分
SingleOutputStreamOperator<Order> mainDataStream = sourceStream.process(new ProcessFunction<Order, Order>(){
@Override
public void processElement(Order value, Context ctx, Collector<Order> out) throws Exception {
//这句代码的含义是把数据发送到常规的流中，也就是mainDataStream中去，发送的数据是全量的数据
//如果不需要全量的数据，可以不进行发送，那么mainDataStream中也就没有数据
out.collect(value);
double amt = value.amount;//获取订单里的金额
if(amt < 200){
ctx.output(outputTag1, value);//把金额小于200的数据发送到旁路流1中
} else if(amt >= 200 && amt < 500){
ctx.output(outputTag2, value);//把金额大于200、小于500的数据发送到旁路流2中
} else {
ctx.output(outputTag3, JSON.toJSONString(value));//把金额大于500的数据发送到旁路流2中,并且转成JSON字符串
}
}
});
三、获取数据
首先mainDataStream中的数据是常规数据，也就是 out.collect进去的数据，可以直接使用
获取旁路数据
DataStream<KfkSourceData> sideOutput1Stream = mainDataStream.getSideOutput(outputTag1);//获取旁路1的数据
DataStream<KfkSourceData> sideOutput2Stream = mainDataStream.getSideOutput(outputTag2);//获取旁路2的数据
DataStream<String> sideOutput3Stream = mainDataStream.getSideOutput(outputTag3);//获取旁路3的数据
注：
原先flink提供了split来进行流的拆分，后来改为扩展性更强的 side output。split已经废弃。
