使用旁路输出(side output)来拆分和复制流

  咱们在处理数据的时候，有时候想对不一样状况的数据进行不一样的处理，那么就须要把流进行拆分或者复制。
若是是使用filter来进行拆分，也能知足咱们的需求，但每次筛选都要保留整个流，而后遍历整个流，显然很浪费性能，假如可以在一个流了屡次输出就行了，flink的旁路输出则提供了这样的功能。

如何使用旁路输出呢

1、要定义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));

2、使用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字符串
                }
            }
        });

3、获取数据
首先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已经废弃。

做者：和平菌
连接：https://www.jianshu.com/p/c009b2a20a83
来源：简书
著做权归做者全部。商业转载请联系做者得到受权，非商业转载请注明出处。