Flink Flink中的分流

本文介绍: 所谓“分流”，就是将一条数据流拆分成完全独立的两条、甚至多条流。也就是基于一个DataStream，定义一些筛选条件，将符合条件的数据拣选出来放到对应的流里。

一、什么是分流

所谓“分流”，就是将一条数据流拆分成完全独立的两条、甚至多条流。也就是基于一个DataStream，定义一些筛选条件，将符合条件的数据拣选出来放到对应的流里。
在这里插入图片描述

二、基于filter算子的简单实现 分流

其实根据条件筛选数据的需求，本身非常容易实现：只要针对同一条流多次独立调用.filter()方法进行筛选，就可以得到拆分之后的流了。
案例需求：读取一个整数数字流，将数据流划分为奇数流和偶数流。

package com.flink.DataStream.SplitStream;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkSplitStreamByFilter {
    public static void main(String[] args) throws Exception {
        //TODO 创建Flink上下文执行环境
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment
                .createLocalEnvironmentWithWebUI(new Configuration().set(RestOptions.BIND_PORT, "8081"));
        //.getExecutionEnvironment();
        //TODO 设置全局并行度为2
        streamExecutionEnvironment.setParallelism(2);
        DataStreamSource<String> dataStreamSource = streamExecutionEnvironment.socketTextStream("localhost", 8888);
        //TODO 先将输入流转为Integer类型
        SingleOutputStreamOperator<Integer> mapResult = dataStreamSource.map((input) -> {
            int i = Integer.parseInt(input);
            return i;
        });
        //TODO 使用匿名函数分流偶数流
        SingleOutputStreamOperator<Integer> ds1 = mapResult.filter(new FilterFunction<Integer>() {
            @Override
            public boolean filter(Integer a) throws Exception {
                return a % 2 == 0;
            }
        });
        //TODO 使用lamda表达式分流奇数流
        SingleOutputStreamOperator<Integer> ds2 = mapResult.filter((a) -> a % 2 == 1);
        ds1.print("偶数流");
        ds2.print("奇数流");
        streamExecutionEnvironment.execute();
    }
}

执行结果

奇数流:1> 1
偶数流:2> 2
偶数流:1> 2
偶数流:2> 4
奇数流:1> 3
奇数流:2> 1

Process finished with exit code 130 (interrupted by signal 2: SIGINT)

这种实现非常简单，但代码显得有些冗余——我们的处理逻辑对拆分出的三条流其实是一样的，却重复写了三次。而且这段代码背后的含义，是将原始数据流 stream 复制三份，然后对每一份分别做筛选；这明显是不够高效的。我们自然想到，能不能不用复制流，直接用一个算子就把它们都拆分开呢？

三、使用测输出流

关于处理函数中侧输出流的用法，我们已经在 7.5 节做了详细介绍。简单来说，只需要调用上下文 ctx 的.output()方法，就可以输出任意类型的数据了。而侧输出流的标记和提取，都离不开一个“输出标签”（OutputTag），指定了侧输出流的 id 和类型。

package com.flink.DataStream.SplitStream;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SplitStreamByOutputTag {
    public static void main(String[] args) throws Exception {
        //TODO 创建Flink上下文环境
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 设置并行度为1
        streamExecutionEnvironment.setParallelism(1);
        //TODO Source
        DataStreamSource<String> dataStreamSource = streamExecutionEnvironment.socketTextStream("localhost", 8888);
        //TODO Transform
        SingleOutputStreamOperator<Object> outputStreamOperator = dataStreamSource.map(new MapFunction<String, Object>() {
            @Override
            public Object map(String input) throws Exception {
                //将socket输入的string转为Int类型
                return Integer.parseInt(input);
            }
        });
        //在main函数中new2个输出标签
        OutputTag<Integer> ji = new OutputTag<Integer>("ji", Types.INT){};
        OutputTag<Integer> ou = new OutputTag<Integer>("ou", Types.INT){};
        //调用底层算子process
        SingleOutputStreamOperator<Object> output0 = outputStreamOperator.process(new ProcessFunction<Object, Object>() {
            @Override
            public void processElement(Object value, ProcessFunction<Object, Object>.Context context, Collector<Object> collector) throws Exception {
                int i = Integer.parseInt(value.toString());
                if (i < 0) {
                    //如果小于0就侧输出流为负数
                    context.output(ou, i);
                } else if (i  > 10) {
                    //如果大于10就侧输出流为异常
                    context.output(ji, i);
                } else {
                    //其他视为正常值流入主流
                    collector.collect(i);
                }
            }
        });
        //TODO Sink
        output0.print("正常");
        DataStream<Integer> output1 = output0.getSideOutput(ji);
        DataStream<Integer> output2 = output0.getSideOutput(ou);

        output1.printToErr("负数");
        output2.printToErr("异常");

        //TODO 执行
        streamExecutionEnvironment.execute();

    }
}