Spark 共享变量

ゝ一世哀愁。 2022-06-12 02:41 177阅读 0赞

Spark中有两种类型的共享变量：一个是累加器accumulator、一个是广播变量broadcast variable。

> 累加器：用来对信息进行聚合  
> 广播变量：用来高效分发较大的对象

--------------------

## 一. 累加器 ##

累加器的一个常见用途是在调试时对作业执行过程中的事件进行计数，如：统计日志中空行数、统计错误行数等。

### 用法： ###

1.  通过调用JavaSparkContext.accumulator(initivalValue)方法，创建出存有初始值的累加器。返回值为org.apache.spark.Accumulator\[T\]对象，其中T是初始值initialValue的类型
2.  Spark闭包里的执行器代码可以使用累加器的+=方法增加累加器的值
3.  驱动器程序可以调用累加器的value属性来访问累加器的值

### 实例： ###

使用累加器统计data.txt文件中的空行数。

public static void main(String args[]) throws InterruptedException {
            SparkConf conf = new SparkConf().setMaster("local").setAppName("accumulator");
            JavaSparkContext context = new JavaSparkContext(conf);
            JavaRDD<String> stringJavaRDD = context.textFile("data.txt");
            final Accumulator<Integer> blankLines = context.accumulator(0);
            JavaRDD<String> stringJavaRDD1 = stringJavaRDD.flatMap(new FlatMapFunction<String, String>() {
                @Override
                public Iterable<String> call(String s) throws Exception {
                    if ("".equals(s)) {
                        blankLines.add(1);
                    }
                    return Arrays.asList(s.split(" "));
                }
            });
            Thread.sleep(5000);
            stringJavaRDD1.foreach(new VoidFunction<String>() {
                @Override
                public void call(String s) throws Exception {
                    System.out.println("s = [" + s + "]");
                }
            });
    
            System.out.println("空行的总数："+blankLines.value());
        }

> **说明**：工作节点上的任务不能访问累加器的值（不能在flatMap中的call方法中打印累加器的值，flatMap中的方法是运行在Excutor中的）。  
> 从这些任务的角度来看，累加器是一个只写变量。在这种模式下，累加器的实现可以更加高效，不需要对每次更新操作进行复杂的通信。

若在工作节点上(call方法中)访问累加器的值：会报Can’t read accumulator value in task的错误，详细如下：

![这里写图片描述][SouthEast]

### 容错性： ###

Spark会自动重新执行失败的或较慢的任务来应对有错误的或者比较慢的机器。例如，如果对某分区执行map()操作的节点失败了，Spark会在另一个节点上重新运行该任务。即使该节点没有崩溃，而只是处理速度比别的节点慢很多，Spark也可以抢占式地在另一个节点上启动一个speculative型的任务副本，如果该任务更早结束就可以直接获取结果。即使没有节点失败，Spark有时也需要重新运行任务来获取缓存中被移除出内存的数据。  
对于要在行动操作中使用的累加器，Spark只会把每个任务对各累加器的修改应用一次。因此如果想要一个无论在失败还是重复计算时都绝对可靠的累加器，我们必须把它放在foreach()这样的行动操作中。

## 二：广播变量 ##

可以让程序高效地向所有工作节点发送一个 较大的只读值，以供一个或多个 Spark 操作使用。

### 用法： ###

1.  通过对一个类型T的对象调用SparkContext.broadcast创建出一个BroadCast\[T\]对象，任何可序列化的类型都可以这么实现。
2.  通过value属性访问该对象的值
3.  变量只会被发到各个节点一次，应作为只读值处理。（修改这个值不会影响到别的节点）

### 实例： ###

读取广播变量的值。

public static void main(String args[]){
            SparkConf conf = new SparkConf().setAppName("broadcast").setMaster("local");
            JavaSparkContext context = new JavaSparkContext(conf);
    
            JavaRDD<String> stringJavaRDD = context.textFile("data.txt");
    
            //定义广播变量
            final Broadcast<String> lsz = context.broadcast("lsz");
    
    
            JavaPairRDD<String, String> stringStringJavaPairRDD = stringJavaRDD.mapToPair(new PairFunction<String, String, String>() {
                @Override
                public Tuple2<String, String> call(String s) throws Exception {
    
                    String value = lsz.getValue();
                    return new Tuple2<String, String>(value, s);
                }
            });
            stringStringJavaPairRDD.foreach(new VoidFunction<Tuple2<String, String>>() {
                @Override
                public void call(Tuple2<String, String> stringStringTuple2) throws Exception {
                    System.out.println(stringStringTuple2);
                }
            });
        }

### 广播的优化： ###

当广播一个比较大的值时，选择既快又好的序列化格式是很重要的，因为如果序列化对象的时间很长或者传送花费的时间太久，这段时间很容易就成为性能瓶颈。尤其是，Spark的Scale和JavaAPI中默认使用的序列化库为Java序列化库，因此它对于除基本类型的数组以外的对象都比较低效的。

> 可以使用spark.serializer属性选择另一个序列化库来优化序列化过程，也可以让数据类型实现自己的序列化方式（对Java对象使用java.io.Externalizable接口实现序列化，或使用reduce()方法为Python的pickle库自定义的序列化）

[SouthEast]: /images/20220612/66b05253ccbe4a8789e3c12e9fbd347a.png