【FLink源码分析】：Accumulator源码分析

雨点打透心脏的1/2处 2023-07-01 14:56 56阅读 0赞

### 文章目录 ###

*  Accumulator源码分析学习
 *   *  Accumulator源码结构
     *   *  Accumulator 结构
         *   *  成员方法
         *  SimpleAccumulator
         *  累加器使用案例
         *   *  code

# Accumulator源码分析学习 #

## Accumulator源码结构 ##

> Flink 的Accumulator是一个累加器，和spark的累加器是一个道理

### Accumulator 结构 ###

public interface Accumulator<V, R extends Serializable> extends Serializable, Cloneable {
    }

#### 成员方法 ####

*  add
 *  getLocalValue
 *  resetLocal
 *  merge
 *  clone

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-XpfrjeSO-1579707620504)(9C13F9BDC6D04C1488448A5F237CE6C7)\]][img-XpfrjeSO-1579707620504_9C13F9BDC6D04C1488448A5F237CE6C7]

### SimpleAccumulator ###

源码中SimpleAccumulator接口继承了累加器，供上层的其他累加器实现这个接口，然后实现具体逻辑

@Public
    public interface SimpleAccumulator<T extends Serializable> extends Accumulator<T,T> {}

1.  `IntCounter，LongCounter，DoubleCounter`：允许将 TaskManager 发送的 int，long，double 值汇总在一起
2.  `LongMaximum，LongMinimum，IntMaximum，IntMinimum，DoubleMaximum，DoubleMinimum`：累加器，用于确定不同类型的最大值和最小值

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70]

> 源码中可以看出有很多具体的不同功能的实现类都实现了这个接口

### 累加器使用案例 ###

#### code ####

lines.flatMap(new RichFlatMapFunction<String, Tuple2<String, Integer>>() {
    
        //创建一个累加器
        private IntCounter linesNum = new IntCounter();
    
        @Override
        public void open(Configuration parameters) throws Exception {
            //注册一个累加器
            getRuntimeContext().addAccumulator("linesNum", linesNum);
        }
    
        @Override
        public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] words = line.split("\\W+");
            for (String word : words) {
                out.collect(new Tuple2<>(word, 1));
            }
            
            // 处理每一行数据后 linesNum 递增
            linesNum.add(1);
        }
    })
    .groupBy(0)
    .sum(1)
    .print();
    
    //获取累加器结果
    int linesNum = env.getLastJobExecutionResult().getAccumulatorResult("linesNum");
    System.out.println(linesNum);

[img-XpfrjeSO-1579707620504_9C13F9BDC6D04C1488448A5F237CE6C7]: https://img-blog.csdnimg.cn/20200122234057300.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020012223411413.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ==,size_16,color_FFFFFF,t_70