Flink会自动检查AggregateFunction的状态吗?如何使用AggregatingStateDescriptor？

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class AggregatingStateExample {

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

        // 创建一个包含两个字段的DataStream
        DataStream<Tuple2<String, Long>> input = env.fromElements(
                Tuple2.of("key", 1L),
                Tuple2.of("key", 2L),
                Tuple2.of("key", 3L)
        );

        // 创建一个AggregatingStateDescriptor，指定状态名称、数据类型和聚合函数
        AggregatingStateDescriptor<Tuple2<String, Long>, AverageAccumulator, Double> descriptor =
                new AggregatingStateDescriptor<>(
                        "average",
                        new AverageAggregateFunction(),
                        Double.class
                );

        // 将状态添加到DataStream上
        DataStream<Double> result = input.keyBy(0)
                .flatMap(new AverageAggregator(descriptor));

        result.print();

        env.execute("AggregatingStateExample");
    }

    // 自定义聚合函数
    public static class AverageAggregateFunction implements AggregateFunction<Tuple2<String, Long>, AverageAccumulator, Double> {

        @Override
        public AverageAccumulator createAccumulator() {
            return new AverageAccumulator();
        }

        @Override
        public AverageAccumulator add(Tuple2<String, Long> value, AverageAccumulator accumulator) {
            accumulator.sum += value.f1;
            accumulator.count++;
            return accumulator;
        }

        @Override
        public Double getResult(AverageAccumulator accumulator) {
            return accumulator.sum / accumulator.count;
        }

        @Override
        public AverageAccumulator merge(AverageAccumulator a, AverageAccumulator b) {
            a.sum += b.sum;
            a.count += b.count;
            return a;
        }
    }

    // 自定义累加器
    public static class AverageAccumulator {
        public long sum;
        public long count;
    }

    // 自定义FlatMapFunction，用于访问AggregatingState
    public static class AverageAggregator extends RichFlatMapFunction<Tuple2<String, Long>, Double> {

        private final AggregatingStateDescriptor<Tuple2<String, Long>, AverageAccumulator, Double> descriptor;
        private AggregatingState<Tuple2<String, Long>, Double> state;

        public AverageAggregator(AggregatingStateDescriptor<Tuple2<String, Long>, AverageAccumulator, Double> descriptor) {
            this.descriptor = descriptor;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            // 获取AggregatingState
            state = getRuntimeContext().getAggregatingState(descriptor);
        }

        @Override
        public void flatMap(Tuple2<String, Long> value, Collector<Double> out) throws Exception {
            // 更新状态
            state.add(value);
            // 获取聚合结果
            out.collect(state.get());
        }
    }
}