Flink 累加器实例

雨点打透心脏的1/2处 2021-09-24 11:38 440阅读 0赞

**每个并行的任务实例下有各自的累加器，获取最终值时会把各个并行任务的累加器的值求和。（一个并行度对应一个累加器）**

版本: flink1.9.2    java1.8

**map并行度1:**

package Counter;
    
    import org.apache.flink.api.common.JobExecutionResult;
    import org.apache.flink.api.common.accumulators.IntCounter;
    import org.apache.flink.api.common.functions.RichMapFunction;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @Author you guess
     * @Date 2020/6/25 12:30
     * @Version 1.0
     * @Desc
     */
    public class CounterTest1 {
    
        public static void main(String[] args) throws Exception {
            //获取执行环境
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //数据来源
            DataStream<String> source1 = env.fromElements("a", "b", "c");
    
            //operate
            source1.map(new RichMapFunction<String, String>() {
    
                //第一步：定义累加器
                IntCounter numLines = new IntCounter();
    
                @Override
                public void open(Configuration parameters) throws Exception {
    
                    //第二步：注册累加器
                    getRuntimeContext().addAccumulator("num-lines", numLines);
                }
    
                @Override
                public String map(String s) throws Exception {
                    //第三步：累加
                    numLines.add(1);
                    System.out.println("LocalValue:" + numLines.getLocalValue()+",线程id#"+Thread.currentThread().getId());
                    return s;
                }
    
            }).setParallelism(1);
    
            //数据去向
            source1.print();
    
            //执行
            JobExecutionResult exe1 = env.execute("Flink CounterTest1 Demo");
    
            //第四步：结束后输出总量；如果不需要结束后持久化，可以省去，因为在flinkUI中可以看到
            String total = exe1.getAccumulatorResult("num-lines").toString();
            System.out.println("total:" + total);
        }//main
    
    }
    /*
    输出：
    LocalValue:1,线程id#79
    LocalValue:2,线程id#79
    LocalValue:3,线程id#79
    1> c
    11> a
    12> b
    total:3
    */

**map并行度不设置（默认是12）:**

package Counter;
    
    import org.apache.flink.api.common.JobExecutionResult;
    import org.apache.flink.api.common.accumulators.IntCounter;
    import org.apache.flink.api.common.functions.RichMapFunction;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @Author you guess
     * @Date 2020/6/25 12:30
     * @Version 1.0
     * @Desc
     */
    public class CounterTest1 {
    
        public static void main(String[] args) throws Exception {
            //获取执行环境
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //数据来源
            DataStream<String> source1 = env.fromElements("a", "b", "c");
    
            //operate
            source1.map(new RichMapFunction<String, String>() {
    
                //第一步：定义累加器
                IntCounter numLines = new IntCounter();
    
                @Override
                public void open(Configuration parameters) throws Exception {
    
                    //第二步：注册累加器
                    getRuntimeContext().addAccumulator("num-lines", numLines);
                }
    
                @Override
                public String map(String s) throws Exception {
                    //第三步：累加
                    numLines.add(1);
                    System.out.println("LocalValue:" + numLines.getLocalValue()+",线程id#"+Thread.currentThread().getId());
                    return s;
                }
    
            });
    
            //数据去向
            source1.print();
    
            //执行
            JobExecutionResult exe1 = env.execute("Flink CounterTest1 Demo");
    
            //第四步：结束后输出总量；如果不需要结束后持久化，可以省去，因为在flinkUI中可以看到
            String total = exe1.getAccumulatorResult("num-lines").toString();
            System.out.println("total:" + total);
        }//main
    
    }
    /*
    输出：
    7> b
    LocalValue:1,线程id#76
    LocalValue:1,线程id#75
    6> a
    LocalValue:1,线程id#74
    8> c
    total:3
    */