spark累加器、广播变量

一言以蔽之：

累加器就是只写变量通常就是做事件统计用的因为rdd是在不同的excutor去执行的你在不同excutor中累加的结果没办法汇总到一起这个时候就需要累加器来帮忙完成

广播变量是只读变量正常的话我们在driver定义一个变量需要序列化才能在excutor端使用而且是每个task都需要传输一次这样如果我们定义的对象很大的话就会产生大量的IO 如果你把这个大对象定义成广播变量的话我们只需要每个excutor发送一份就可以如果task需要时只需要从excutor拉取就可以了。可以减轻集群driver和executor间的通信压力，节省集群资源。

Spark两种共享变量：广播变量（broadcast variable）与累加器（accumulator），广播变量常用来高效分发较大的对象，而累加器用来对信息进行聚合。

共享变量出现的原因：通常在向 Spark 传递函数时，比如使用map或reduce传条件或变量时，在driver端定义变量，但是集群中运行的每个任务都会得到这些变量的一份新的副本，更新这些副本的值driver端的对应变量并不会随之更新。Spark 的两个共享变量，广播变量与累加器分别为变量提供广播与聚合功能，突破了变量不能共享的限制。

2、广播变量的使用原则

不能将RDD用做变量广播出去，RDD是不存储数据的，可以将RDD的结果广播出去。
广播变量只能在Driver端定义，不能在Executor端定义，Executor端只能使用。
广播变量的值只能在Driver端修改，在Executor端不能修改

3、广播变量使用方法

(1) 通过SparkContext.Broadcast[T] 创建一个变量v，并进行广播，广播变量以序列化形式缓存。

如scala方式：val broadCast = sc.broadcast(T) 对T进行广播，T可以是任何能被序列化的类型

(2) 通过 broadCast.value 属性访问该对象的值，而不能直接访问T

(3) 当要更新广播变量时候，通过broadCast.unpersist()方法清除广播变量，之后可重新广播
4、广播变量使用场景

日常工作中对来访url、ip等过滤业务，就可以使用广播变量

112.168.102.71
19.18.172.75
12.16.72.20
100.20.13.4
8.8.8.8
102.168.17.205
202.102.12.74
114.114.114.114

object FilterIP {
   def main(args: Array[String]): Unit = {
      val conf = new SparkConf().setAppName("broadcast");
      conf.setMaster("local");
      
      val sc = new SparkContext(conf);
      
      val lineRdd = sc.textFile("./ip_list.txt");//外部数据源,从别的数据源读取
      
      //使用广播变量时，driver第一次向executor发送task时候，发送blackList，缓存到blockmanager，以后不会再发送
      val blackList = List[String]("8.8.8.8","114.114.114.114");//过滤黑名单ip,从别的数据源读取
      //在driver端进行广播blackList
      val broadCast = sc.broadcast(blackList);
      
      //在executor端用broadCast.value获取blackList的值
      val filterRdd =lineRdd.filter(ip =>{ !broadCast.value.contains(ip)})
       
      filterRdd.foreach(ip =>{
         println(ip)
         //处理其他业务
      })     
   }
}

112.168.102.71
19.18.172.75
12.16.72.20
100.20.13.4
102.168.17.205
202.102.12.74

使用spark广播变量时候，在输出结果是看不到有什么变化，但这种变化是内在的，可以减轻集群driver和executor间的通信压力，节省集群资源。

Spark累加器

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
 
object FilterIP {
   def main(args: Array[String]): Unit = {
      val conf = new SparkConf().setAppName("broadcast");
      conf.setMaster("local");
      
      val sc = new SparkContext(conf);
      
      val lineRdd = sc.textFile("./ip_list.txt");//外部数据源,从别的数据源读取
      var sum = 0;
      val blackList = List[String]("8.8.8.8","114.114.114.114");//过滤黑名单ip,从别的数据源读取
      //在driver端进行广播blackList
      val broadCast = sc.broadcast(blackList);
      val filterRdd =lineRdd.filter(ip =>{ broadCast.value.contains(ip)})
      filterRdd.foreach(ip =>{
        sum+=1;
        println("executor sum="+sum)
      })
      //打印出非法ip的访问数量
      println("driver sum="+sum)
   }
}

输出：

executor sum=1
executor sum=2
 
driver sum=0

可以看到在driver端sum仍然是0，这并不是我们想要的结果。原来sum在各个节点的executor中累加的同时，driver端的sum最后并不会更新，导致sum最终仍然是0。

spark的累加器就是解决此类问题而出现的，其提供了聚合功能，把各个节点上的executor对变量的累加结果聚合到driver端，最终统计出我们想要的结果。spark的累加器充分彰显的分布式计算的特性。
2、累加器使用原则

累加器在Driver端定义赋初始值，累加器只能在Driver端读取，在Excutor端更新。

3、累加器的使用方法

(1) 通过sparkContext.longAccumulator()或sparkContext.doubleAccumulator()来累积long或double类型的值来创建数字累加器

如scala方式：var accumulator=sc.longAccumulator("accumulator");

(2) 在executor端通过accumulator.add(1)进行累加后并回传到driver

4、累加器的使用场景

如对非法来访Ip的统计

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
 
object FilterIP {
   def main(args: Array[String]): Unit = {
     val conf = new SparkConf().setAppName("broadcast");
      conf.setMaster("local");
      
      val sc = new SparkContext(conf);
      
      val lineRdd = sc.textFile("./ip_list.txt");//外部数据源,从别的数据源读取
      //spark2.0.0中sc.accumulator(0)不被推荐，用如下方式初始值，默认0
      val accumulator=sc.longAccumulator("accumulator");
      
      val blackList = List[String]("8.8.8.8","114.114.114.114");//过滤黑名单ip,从别的数据源读取
      //在driver端进行广播blackList
      val broadCast = sc.broadcast(blackList);
      val filterRdd =lineRdd.filter(ip =>{ broadCast.value.contains(ip)})
      filterRdd.foreach(ip =>{
        accumulator.add(1);
      })
      //打印出非法ip的访问数量
      println("driver sum="+accumulator.value)
   }
}

输出：

driver sum=2

————————————————
版权声明：本文为CSDN博主「LiryZlian」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/lzxlfly/article/details/86366722

import org.apache.spark.SparkConfimport org.apache.spark.SparkContext object FilterIP { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("broadcast"); conf.setMaster("local"); val sc = new SparkContext(conf); val lineRdd = sc.textFile("./ip_list.txt");//外部数据源,从别的数据源读取 //spark2.0.0中sc.accumulator(0)不被推荐，用如下方式初始值，默认0 val accumulator=sc.longAccumulator("accumulator"); val blackList = List[String]("8.8.8.8","114.114.114.114");//过滤黑名单ip,从别的数据源读取 //在driver端进行广播blackList val broadCast = sc.broadcast(blackList); val filterRdd =lineRdd.filter(ip =>{ broadCast.value.contains(ip)}) filterRdd.foreach(ip =>{ accumulator.add(1); }) //打印出非法ip的访问数量 println("driver sum="+accumulator.value) }}