AccTest.java

public class AccTest {

    public static final String ACC_NAME1 = "accName";

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

        DataSource<Integer> source = env.fromCollection(Lists.newArrayList(1, 2, 3, 4, 5));


        source.map(new AccMap()).setParallelism(1)
              .map(new AccMap()).setParallelism(2)
              .print();

        source.map(new AccMap()).setParallelism(1);

        System.out.println(env.getLastJobExecutionResult()
                                                       .getAllAccumulatorResults());


    }

    public static class AccMap extends RichMapFunction<Integer, Integer> {

        private LongCounter counter = new LongCounter();

        @Override
        public void open(Configuration parameters) throws Exception {
            getRuntimeContext().addAccumulator(ACC_NAME1, counter);
        }

        @Override
        public Integer map(Integer integer) throws Exception {
            return integer;
        }
    }
}