在Flink中，如何将累加值转换为flink中的增量值，然后按多个键聚合

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

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

        // 创建一个DataStream，包含键值对（key，value）
        DataStream<Tuple2<String, Integer>> input = env.fromElements(
                new Tuple2<>("key1", 1),
                new Tuple2<>("key2", 2),
                new Tuple2<>("key1", 3),
                new Tuple2<>("key2", 4)
        );

        // 按键进行分组
        DataStream<Tuple2<String, Integer>> grouped = input.keyBy(0);

        // 定义一个滚动窗口，窗口大小为5秒
        DataStream<Tuple2<String, Integer>> windowed = grouped.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                // 使用sum聚合函数将累加值转换为增量值
                .aggregate(new SumAggregateFunction());

        // 打印结果
        windowed.print();

        env.execute("Flink Aggregation Example");
    }

    public static class SumAggregateFunction implements AggregateFunction<Tuple2<String, Integer>, Integer, Tuple2<String, Integer>> {
        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(Tuple2<String, Integer> value, Integer accumulator) {
            return accumulator + value.f1;
        }

        @Override
        public Tuple2<String, Integer> getResult(Integer accumulator) {
            return new Tuple2<>("sum", accumulator);
        }

        @Override
        public Integer merge(Integer a, Integer b) {
            return a + b;
        }
    }
}