Spark广播变量

什么是广播变量？

在同一个Execute共享同一份计算逻辑的变量

广播变量使用场景

我现在要在在这些内容中过滤java和object-c

过滤内容

• 使用广播变量过滤代码逻辑：

package com.chenzhipeng.spark.examples01;

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.broadcast.Broadcast;

import java.util.ArrayList;
import java.util.List;

/**
 * <p>Title:BroadcastMain</p>
 * <p>Description:广播变量</p>
 * @version V1.0
 * @author ZhiPeng_Chen
 * @date: 2017/12/23
 */
public class BroadcastMain {

    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf();

        sparkConf.setMaster("local").setAppName("BroadcastMain");

        JavaSparkContext context = new JavaSparkContext(sparkConf);

        JavaRDD<String> rdd = context.textFile("examples-01/src/main/resources/names");

        //添加过滤的信息
        List list = new ArrayList<String>();
        list.add("java");
        list.add("object-c");

        //使用广播变量
       final Broadcast<List> broadcast = context.broadcast(list);

        JavaRDD<String> filter = rdd.filter(new Function<String, Boolean>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Boolean call(String v1) throws Exception {
                //过滤集合中的内容
                return !broadcast.value().contains(v1);

            }
        });

        //触发执行后的过滤后结果
        List<String> collect = filter.collect();

        for (String str : collect) {
            System.out.println(str);
        }

        context.close();
        context.stop();

    }

}

执行结果:

过滤后结果

为什么要使用广播变量？

主要用于节约内存的开销。

Driver每次分发任务的时候会把task和计算逻辑的变量发送给Executor，不是使用广播变量会有多份的变量副本。这样会导致消耗大量的内存导致严重的后果。

• 不使用广播变量:

不使用广播变量

• 使用广播变量:

使用广播变量

总结

如果不使用广播变量，task的个数越多越消耗性能，使用广播变量Driver端不管分发有多少个task在同一个Executor只享用同一份变量。

使用注意事项

• 广播变量在Driver端定义
• 广播变量在Execoutor只能读取不能修改
• 广播变量的值只能在Driver端修改