flink sideoutput

flink 输出sideoutput

定义一个OutPutTag，注意一定需要增加{} 

OutputTag<SensorEntity> high = new OutputTag<SensorEntity>("high"){};

public class SideOutputActionTest3 {

    //拆分数据流 温度大于30 和小于30 sideoutput

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);


        DataStreamSource<String> inputSource = env.addSource(new SimpleUdfSource());

        SingleOutputStreamOperator<SensorEntity> map = inputSource.map(line -> {

            String[] str = line.split(",");
            return new SensorEntity(str[0], Long.parseLong(str[1]), Double.valueOf(str[2]));

        });


        // 定义一个sideoutput
        OutputTag<SensorEntity> high = new OutputTag<SensorEntity>("high"){};
        OutputTag<SensorEntity> low = new OutputTag<SensorEntity>("low"){};

        SingleOutputStreamOperator<SensorEntity> processstream = map.process(new ProcessFunction<SensorEntity, SensorEntity>() {
            @Override
            public void processElement(SensorEntity sensorEntity, Context context, Collector<SensorEntity> collector) throws Exception {
                if (sensorEntity.getTmpeature() > 40) {

                    context.output(high, sensorEntity);
                } else {
                    context.output(low, sensorEntity);

                }
            }
        });


        DataStream<SensorEntity> highSideOutput = processstream.getSideOutput(high);
        DataStream<SensorEntity> lowSideOutput = processstream.getSideOutput(low);

        highSideOutput.print("high");
        lowSideOutput.print("low");
        map.print("all");
}
}