Flink SideOutput 数据分流处理

大数据学习之路

不积跬步，无以至千里。不积小流，无以成江海。——荀子

为什么要数据分流

通常数据流中包含多种类型数据，每种类型数据需要进行不同的分析。因此我们不得不对数据进行分流。

传统的方式我们可能需要对数据流进行多次遍历或者复制,本文学习Flink如何对数据流进行分流处理,这就是文章主角SideOutput

从一个简单例子讲起

下面代码例子，将原始的Integer 数据流 分流为奇数、偶数; 通过getSideOutput 得到两个数据流，一个只包含奇数的数据流，一个只包含偶数的数据流，然后对分流后的数据流进行打印。

Flink SideOutput使用流程

• 1.定义不同分流的OutputTag

OutputTag<Integer> evenTag = new OutputTag<Integer>("even"){};
OutputTag<Integer> oddTag = new OutputTag<Integer>("odd"){};

• 2.根据不同的条件将数据输出到不同的 side output

if(value % 2 == 0){
    ctx.output(evenTag,value);
}else {
    ctx.output(oddTag,value);
}

• 3.使用 SingleOutputStreamOperator#getSideOutput(OutputTagsideOutputTag)方法获取对应的数据流

SideOutputDataStream<Integer> evenStream = passThroughtStream.getSideOutput(evenTag);

SideOutputDataStream<Integer> oddStream = passThroughtStream.getSideOutput(oddTag);

• 4.对不同数据流进行处理 本案例只是简单的打印

evenStream.print("Even Numbers: ");
// 打印奇数流
oddStream.print("Odd Numbers: ");

完整的代码

 public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建一个简单的整数数据流
        DataStream<Integer> dataStream = env.fromElements(1, 2, 3, 4, 5, 6);

        OutputTag<Integer> evenTag = new OutputTag<Integer>("even"){};
        OutputTag<Integer> oddTag = new OutputTag<Integer>("odd"){};

        SingleOutputStreamOperator<Integer> passThroughtStream = dataStream.process(new ProcessFunction<Integer, Integer>() {
            @Override
            public void processElement(Integer value, ProcessFunction<Integer, Integer>.Context ctx, Collector<Integer> out) throws Exception {

                if(value % 2 == 0){
                    ctx.output(evenTag,value);
                }else {
                    ctx.output(oddTag,value);
                }

            }
        });


        SideOutputDataStream<Integer> evenStream = passThroughtStream.getSideOutput(evenTag);

        SideOutputDataStream<Integer> oddStream = passThroughtStream.getSideOutput(oddTag);
        // 打印偶数流
        evenStream.print("Even Numbers: ");
        // 打印奇数流
        oddStream.print("Odd Numbers: ");

        // 执行任务
        env.execute();
    }

总结

• 分流处理

  对于一个复杂的数据流可能包含多种属性，我们需要对这个数据流进行多维度分析，因此我们可能需要根据数据的某些属性将其分流到不同的处理路径。

  例如，一个数字流我们可以按照 正数、负数、奇数、偶数、质数等进行分流处理。

  使用Side Output可以轻松实现这种分流，而无需多次复制数据流，从而提高处理效率。

• 异常值捕获：实际的业务数据中通常包含一些不合法的数据（如空值、数据过大的异常值等）。使用Side Output可以将这些异常数据分流，然后对异常数据进行处理分析。

  前面文章中 Watermark对延迟数据处理，其中一种方式就是输出到Side Output 。

Flink 分流机制避免了多次复制数据流，从而减少系统的资源消耗，提高了处理效率。