[大数据]Spark共享变量

默认情况下，如果在一个算子函数中使用到了某个外部的变量，那么这个变量的值会被拷贝到每个task中。此时每个task只能操作自己的那份变量副本。如果多个task想要共享某个变量，那么这种方式是做不到的。

Spark为此提供了两种共享变量：

• 一种是Broadcast Variable(广播变量)
• 另一种是Accumulator(累加变量)

Broadcast Variable

Broadcast Variable会将使用到的变量，仅仅为每个节点拷贝一份，而不会为每个task都拷贝一份副本，因此其最大的作用，就是减少变量到各个节点的网络传输消耗，以及在各个节点上的内存消耗
通过调用SparkContext的broadcast()方法，针对某个变量创建广播变量

注意:广播变量，是只读的
然后在算子函数内，使用到广播变量时，每个节点只会拷贝一份副本。可以使用广播变量的value()方法 获取值。

接下来看一个图深入理解一下

先看左边的代码
这个是一个咱们经常使用的map算子的代码，map算子中执行对每一个元素乘以一个固定变量的操作，此时这个固定的变量属于外部变量。
默认情况下，算子函数内，使用到的外部变量，会被拷贝到执行这个算子的每一个task中

看图中间的MapTask，这些都是map算子产生的task，也就是说这个外部变量会被拷贝到每一个task中。
如果这个外部变量是一个集合，集合中有上亿条数据，这个网络传输就会很耗时，而且在每个task上，占用的内存空间，也会很大

如果算子函数中使用的外部变量，是广播变量的话，那么每个变量只会拷贝一份到每个节点上。节点上所有的task都会共享这一份变量，就可以减少网络传输消耗的时间，以及减少内存占用了。

大家可以想象一个极端情况，如果map算子有10个task，恰好这10个task还都在一个worker节点上，那么这个时候，map算子使用的外部变量就会在这个worker节点上保存10份，这样就很占用内存了。

下面我们来具体使用一下这个广播变量

Scala版本

import org.apache.spark.{SparkConf, SparkContext}


object BoradcastOpScala {

  	def main(args: Array[String]): Unit = {
	    val conf = new SparkConf()
	    conf.setAppName("BoradcastOpScala")
	      .setMaster("local")
	    val sc = new SparkContext(conf)
	
	    val dataRDD = sc.parallelize(Array(1,2,3,4,5))
	    val varable = 2
	    //dataRDD.map(_ * varable)
	    //1：定义广播变量
	    val varableBroadcast = sc.broadcast(varable)
	
	    //2：使用广播变量，调用其value方法
	    dataRDD.map(_ * varableBroadcast.value).foreach(println(_))
	
	    sc.stop()
  	}
}

Java版本

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 java.util.Arrays;


public class BroadcastOpJava {

    public static void main(String[] args) {
        //创建JavaSparkContext
        SparkConf conf = new SparkConf();
        conf.setAppName("BroadcastOpJava")
                .setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
        int varable = 2;
        //1：定义广播变量
        Broadcast<Integer> varableBroadcast = sc.broadcast(varable);

        //2：使用广播变量
        dataRDD.map(new Function<Integer, Integer>() {
            @Override
            public Integer call(Integer i1) throws Exception {
                return i1 * varableBroadcast.value();
            }
        }).foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer i) throws Exception {
                System.out.println(i);
            }
        });

        sc.stop();
    }
}

Accumulator

Spark提供的Accumulator，主要用于多个节点对一个变量进行共享性的操作。

正常情况下在Spark的任务中，由于一个算子可能会产生多个task并行执行，所以在这个算子内部执行的 聚合计算都是局部的，想要实现多个task进行全局聚合计算，此时需要使用到Accumulator这个共享的累加变量。

注意:Accumulator只提供了累加的功能。在task只能对Accumulator进行累加操作，不能读取它的值。 只有在Driver进程中才可以读取Accumulator的值。

下面我们来写一个案例

Scala版本

import org.apache.spark.{SparkConf, SparkContext}

object AccumulatorOpScala {
  	def main(args: Array[String]): Unit = {
	    val conf = new SparkConf()
	    conf.setAppName("AccumulatorOpScala")
	      .setMaster("local")
	    val sc = new SparkContext(conf)
	    val dataRDD = sc.parallelize(Array(1,2,3,4,5))
	
	    //所以此时想要实现累加操作就需要使用累加变量了
	    //1：定义累加变量
	    val sumAccumulator = sc.longAccumulator
	
	    //2：使用累加变量
	    dataRDD.foreach(num=>sumAccumulator.add(num))
	
	    //注意：只能在Driver进程中获取累加变量的结果
	    println(sumAccumulator.value)
	
	    sc.stop()
  	}

}

Java版本

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.VoidFunction;
import org.apache.spark.util.LongAccumulator;

import java.util.Arrays;


public class AccumulatorOpJava {

    public static void main(String[] args) {
        //创建JavaSparkContext
        SparkConf conf = new SparkConf();
        conf.setAppName("AccumulatorOpJava")
                .setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
        //1：定义累加变量
        LongAccumulator sumAccumulator = sc.sc().longAccumulator();

        //2：使用累加变量
        dataRDD.foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer i) throws Exception {
                sumAccumulator.add(i);
            }
        });

        //获取累加变量的值
        System.out.println(sumAccumulator.value());

        sc.stop();
    }

}