使用旁路輸出(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
來源：簡書
著做權歸做者全部。商業轉載請聯繫做者得到受權，非商業轉載請註明出處。