1、代码逻辑实现

package day02;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @program: Flink_learn
 * @description: 滚动聚合
 * @author: Mr.逗
 * @create: 2021-09-17 10:35
 **/
public class SelfKeyBy {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple2<Integer, Integer>> source = env.fromElements(
                Tuple2.of(1, 2)
                , Tuple2.of(1, 3)
        );
        //1、匿名keyBy+sum
        source.keyBy(v->v.f0).sum(1).print();
        //2、匿名keyBy+ReduceFunction
        source.keyBy(v->v.f0).reduce(new ReduceFunction<Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> reduce(Tuple2<Integer, Integer> v1, Tuple2<Integer, Integer> v2) throws Exception {
                return Tuple2.of(v1.f0,v1.f1+v2.f1);
            }
        }).print();
        //3、匿名keyBy+匿名reduce
        source.keyBy(v->v.f0).reduce((Tuple2<Integer,Integer> v1,Tuple2<Integer,Integer> v2)->
        {
            return Tuple2.of(v1.f0,v1.f1+v2.f1);
        }).print();
        //4、非匿名keyBy+非匿名reduce
        KeyedStream<Tuple2<Integer, Integer>, Integer> keyedStream = source.keyBy(new KeySelector<Tuple2<Integer, Integer>, Integer>() {
            @Override
            public Integer getKey(Tuple2<Integer, Integer> v) throws Exception {
                return v.f0;
            }
        });
        keyedStream.reduce(new ReduceFunction<Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> reduce(Tuple2<Integer, Integer> v1, Tuple2<Integer, Integer> v2) throws Exception {
                return Tuple2.of(v1.f0,v1.f1+v2.f1);
            }
        }).print();
        String name = SelfKeyBy.class.getName();
        env.execute(name);
    }
}

复制

2、结果之展示

6> (1,2)
6> (1,5)
6> (1,2)
6> (1,5)
6> (1,2)
6> (1,2)
6> (1,5)
6> (1,5)
复制