共享变量 Accumulator 实例

原创
2017/07/24 15:41
阅读数 22
package cn.hhb.spark.core;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.Accumulator;

import java.util.Arrays;
import java.util.List;

/**
 * Created by dell on 2017/7/13.
 */
public class AccumulatorVariable {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("AccumulatorVariable")
                .setMaster("local")
                .set("spark.testing.memory", "2147480000");

        JavaSparkContext sc = new JavaSparkContext(conf);

        /**
         * 创建Accumulator变量,需要调用SparkContext的accumulator方法
         */
        final Accumulator<Integer> sum = sc.accumulator(0);

        List<Integer> numberList = Arrays.asList(1,2,3,4,5);
        JavaRDD<Integer> numbers = sc.parallelize(numberList);

        numbers.foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer integer) throws Exception {
                // 在函数内部就可以对accumulator变量调用add方法累加值
                sum.add(integer);
            }
        });

        // 在driver程序中,可以调用accumulator的value()方法,获取其值
        System.out.println(sum.value());

        sc.close();
    }
}

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部