有没有在Flink中使用直方图累加器的例子

import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.SimpleAccumulator;
importimport org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

import java.util.ArrayList;
import java.util.List;

public class HistogramAccumulatorExample {

    public static void main(String[] args) throws Exception {
        // 创建一个Flink执行环境

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 创建一个数据集

        DataSet<Integer> input = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        // 使用map函数来处理数据集，并在其中使用直方图累加器

        DataSet<Integer> result = input.map(new RichMapFunction<Integer, Integer>() {

            private HistogramAccumulator histogramAccumulator;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                histogramAccumulator = new HistogramAccumulator();
                getRuntimeContext().addAccumulator("histogram", histogramAccumulator);
            }

            @Override
            public Integer map(Integer value) throws Exception {
                // 将数据添加到直方图累加器中
                histogramAccumulator.add(value);
                return value;
            }
        });

        // 执行任务并获取累加器的结果

        JobExecutionResult jobResult = env.execute("Histogram Accumulator Example");
        List<Integer> histogramResult = jobResult.getAccumulatorResult("histogram");

        // 输出直方图累加器的结果

        System.out.println("Histogram Result:");
        for (Integer value : histogramResult) {
            System.out.println(value);
        }
    }

    // 自定义直方图累加器

    public static class HistogramAccumulator extends SimpleAccumulator<Integer> {

        private List<Integer> histogram;

        public HistogramAccumulator() {
            this.histogram = new ArrayList<>();
        }

        @Override
        public void add(Integer value) {
            histogram.add(value);
        }

        @Override
        public Integer getLocalValue() {
            return histogram.size();
        }

        @Override
        public void merge(Accumulator<Integer, Integer> other) {
            histogram.addAll(((HistogramAccumulator) other).histogram);
        }

        @Override
        public void resetLocal() {
            histogram.clear();
        }

        @Override
        public HistogramAccumulator clone() {
            HistogramAccumulator clone = new HistogramAccumulator();
            clone.histogram.addAll(this.histogram);
            return clone;
        }
    }
}