Flink 侧输出流使用

flink处理数据流时，经常会遇到这样的情况：处理一个数据源时，往往需要将该源中的不同类型的数据做分割（分流）处理，假如使用 filter算子对数据源进行筛选分割的话，势必会造成数据流的多次复制，造成不必要的性能浪费；

flink中的侧输出，就是将数据流进行分割，而不对流进行复制的一种分流机制。flink的侧输出的另一个作用就是对延时迟到的数据进行处理，这样就可以不必丢弃迟到的数据；

简单理解就是，根据业务上的一定规则，将一个源中的数据拆分成不同的流，即主流和侧输出流；

举例来说，源数据流中有一批监控某流水线传感器温度的数据，我们需要将这批数据按照30为一个基准进行拆分，业务上更加关注的是超过30度的数据，因此可以作为主流输出，而低于30度的数据并不想丢弃，因此作为侧输出流，在侧输出流中做后续的处理，下面来看具体的代码演示，

import com.congge.source.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
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 SideOutProcess1 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // socket文本流
        DataStream inputStream = env.socketTextStream("localhost", 7777);
        // 转换成SensorReading类型
        DataStream dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 定义一个OutputTag，用来表示侧输出流低温流
        OutputTag lowTempTag = new OutputTag("lowTemp") {
        };

        // 测试ProcessFunction，自定义侧输出流实现分流操作
        SingleOutputStreamOperator highTempStream = dataStream.process(new ProcessFunction() {
            @Override
            public void processElement(SensorReading value, Context ctx, Collector out) throws Exception {
                // 判断温度，大于30度，高温流输出到主流；小于低温流输出到侧输出流
                if (value.getTemperature() > 30) {
                    out.collect(value);
                } else {
                    ctx.output(lowTempTag, value);
                }
            }
        });

        highTempStream.print("high-temp");
        highTempStream.getSideOutput(lowTempTag).print("low-temp");

        env.execute();
    }

}