一、累加器

分布式共享只写变量。

1.1、实现原理

累加器用来把Executor 端变量信息聚合到Driver 端。在Driver 程序中定义的变量，在Executor 端的每个Task 都会得到这个变量的一份新的副本，每个task 更新这些副本的值后， 传回Driver 端进行merge。

1、 累加器在Driver端定义赋初始值，累加器只能在Driver端读取最后的值，在Excutor端更新。
2、累加器不是一个调优的操作，因为如果不这样做，结果是错的。

1.2、基本使用

-- 基本api
val rdd = sc.makeRDD(List(1,2,3,4,5))
// 声明累加器
var sum = sc.longAccumulator("sum"); 
rdd.foreach(
    num => {
        //使用累加器，此处调用累加器的add方法
        sum.add(num)
    }
)
// 获取累加器的值
println("sum = " + sum.value)

1.3、自定义累加器

// 自定义累加器
// 1. 继承 AccumulatorV2，并设定泛型
// 2. 重写累加器的抽象方法
import org.apache.spark.util.AccumulatorV2
import scala.collection.mutable
//指定泛型参数，IN的类型，OUT的类型
class WordCountAccumulator extends AccumulatorV2[String, mutable.Map[String, Long]]{
  var map : mutable.Map[String, Long] = mutable.Map()
  // 累加器是否为初始状态
  override def isZero: Boolean = {
    map.isEmpty
  }
  // 复制累加器
  override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = {
    new WordCountAccumulator
  }
  // 重置累加器
  override def reset(): Unit = {
    map.clear()
  }
  // 向累加器中增加数据 (In)
  override def add(word: String): Unit = {
    // 查询 map 中是否存在相同的单词
    // 如果有相同的单词，那么单词的数量加 1
    // 如果没有相同的单词，那么在 map 中增加这个单词
    map(word) = map.getOrElse(word, 0L) + 1L
  }
  // Driver合并多个累加器
  override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit =  {
    val map1 = map //当前累加器的累加器变量
    val map2 = other.value //其他累加器的变量
    // 两个 Map 的合并
    map = map1.foldLeft(map2)(
      ( innerMap, kv ) => {
        innerMap(kv._1) = innerMap.getOrElse(kv._1, 0L) + kv._2 
        innerMap
      }
    )
  }
  // 返回累加器的结果 （Out）
  override def value: mutable.Map[String, Long] = map
}
---------------------
使用：
val acc = new WordCountAccumulator()
sc.register(acc,"累加器自定义名字")
// 这样就可以正常使用累加器 acc

1.4、注意实现

少加：转换算子中调用累加器，如果没有行动算子的话，那么不会执行
多加：转换算子中调用累加器，如果有多个行动算子的话，那么会多次执行

二、广播变量

分布式共享只读变量。

2.1、实现原理

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个或多个Spark 操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询表， 广播变量用起来都很顺手。在多个并行操作中使用同一个变量，但是 Spark 会为每个任务分别发送。
由于闭包的数据，都是以Task为单位进行发送的，这样每个Task都可能会包含相同的闭包数据；就会导致同一个Executor包含大量重复数据，从而导致内存爆炸。

而使用广播变量则可以将闭包的数据存放在Executor的JVM内存中，每个Task访问该共享变量即可，从而提高共享效率。

2.2、基本使用

val rdd1 = sc.makeRDD(List( ("a",1), ("b", 2), ("c", 3), ("d", 4) ),4)
val list = List( ("a",4), ("b", 5), ("c", 6), ("d", 7) )
// 声明广播变量
val broadcast: Broadcast[List[(String, Int)]] = sc.broadcast(list)
val resultRDD: RDD[(String, (Int, Int))] = rdd1.map { 
  case (key, num) => {
        var num2 = 0
        // 使用广播变量
        for ((k, v) <- broadcast.value) { 
      if (k == key) {
                num2 = v
            }
        }
        (key, (num, num2))
    }
}