Flink笔记-迭代计算

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

        DataStreamSource source = env.addSource(new FlinkKafkaConsumer010("dwd", new SimpleStringSchema(), KafkaUtils.comsumerProps()));

        SingleOutputStreamOperator> vod = source.map(new MapFunction>() {
            @Override
            public Tuple3 map(String s) throws Exception {
                JSONObject jn = JSON.parseObject(s);
                return new Tuple3<>(jn.getString("userid"), Integer.parseInt(jn.getString("seconds")), Integer.parseInt(jn.getString("seconds")));
            }
        });

        // iterate方法将一个流转换为迭代流
        IterativeStream> iterate = vod.iterate();

        // 迭代条件，满足条件则迭代
        SingleOutputStreamOperator> feedback = iterate.filter(new FilterFunction>() {
            @Override
            public boolean filter(Tuple3 t) throws Exception {
                return t.f1 < 80;
            }
        });

        // 迭代计算逻辑
        SingleOutputStreamOperator> iterateBody = feedback.map(new MapFunction, Tuple3>() {
            @Override
            public Tuple3 map(Tuple3 t) throws Exception {
                return new Tuple3<>(t.f0, t.f1 + 10, t.f2);
            }
        });

        //
        iterate.closeWith(iterateBody);

        // 满足条件则跳出迭代，输出数据
        SingleOutputStreamOperator> result = iterate.filter(new FilterFunction>() {
            @Override
            public boolean filter(Tuple3 t) throws Exception {
                return t.f1 >= 80;
            }
        });

        result.print();

        env.execute();
        
    }
}