SparkCore（五）：累加器和广播变量

蔚落 2022-11-20 10:53 164阅读 0赞

### 文章目录 ###

*  一、累加器
 *  二、广播变量

# 一、累加器 #

**累加器的使用**

累加器（分布式共享只写变量）：

1.  将累加器变量注册到`spark`中
2.  执行计算时，`spark`会将累加器发送到`executor`执行计算
3.  计算完毕后，`executor`会将累加器的计算结果返回到`driver`端。
4.  `driver`端获取到多个累加器的结果，然后两两合并，最后得到累加器的执行结果。

使用累加器前：

val rdd = sc.makeRDD(List(1, 2, 3, 4))
        var sum = 0
        rdd.foreach(t => sum += t)
        println(sum) //0

使用累加器后：

val rdd = sc.makeRDD(List(1, 2, 3, 4))
        //创建累加器
        val sum = sc.longAccumulator("sum")
        rdd.foreach(t => sum.add(t))
        println(sum.value) //10

**自定义累加器**

需求：使用自定义累加器完成`wordcount`

def main(args: Array[String]): Unit = { 
        val sparkConf = new SparkConf().setMaster("local").setAppName("CustomAcc")
        val sc = new SparkContext(sparkConf)
        val rdd = sc.makeRDD(List("hello spark", "scala", "scala spark"))
        //1、创建自定义累加器
        val myAcc = new MyAccumulator
        //2、注册累加器
        sc.register(myAcc)
        //3、使用累加器
        rdd.flatMap(t => t.split(" ")).foreach(
          word => { 
            myAcc.add(word)
          }
        )
        println(myAcc.value)// Map(spark -> 2, scala -> 2, hello -> 1)
      }
      
      /** * 自定义累加器定义 * IN：输入；OUT：输出 */
      class MyAccumulator extends AccumulatorV2[String, mutable.Map[String, Int]] { 
    
        var wordCountMap = mutable.Map[String, Int]()
    
        //判断累加器是否为空
        override def isZero: Boolean = wordCountMap.isEmpty
    
        //复制累加器
        override def copy(): AccumulatorV2[String, mutable.Map[String, Int]] = new MyAccumulator
    
        //重置累加器
        override def reset(): Unit = wordCountMap.clear()
    
        //向累加器中添加值
        override def add(v: String): Unit = { 
          wordCountMap.update(v, wordCountMap.getOrElse(v, 0) + 1)
        }
    
        //合并累加器
        override def merge(other: AccumulatorV2[String, mutable.Map[String, Int]]): Unit = { 
          val map1 = wordCountMap
          val map2 = other.value
    
          wordCountMap = map1.foldLeft(map2)((map, kv) => { 
            map(kv._1) = map.getOrElse(kv._1, 0) + kv._2
            map
          })
        }
    
        // 返回累加器的值
        override def value: mutable.Map[String, Int] = wordCountMap
      }

合并两个`map`方法`foldLeft`说明：  
![在这里插入图片描述][20201028200354691.png]

# 二、广播变量 #

`Spark`中计算数据时，使用的闭包操作。会导致变量被不同的`task`所使用，那么会传递到`executor`多份。这样的话。`executor`内存中就存在大量的数据冗余，性能会下降的比较厉害。

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个或多个`Spark`操作使用。比如，如果你的应用需要向所有节点发送一个较大的**只读查询表**，广播变量用起来都很顺手。在多个并行操作中使用同一个变量，但是`Spark`会为每个任务分别发送。

val sparkConf = new SparkConf().setMaster("local").setAppName("BroadcastTest3")
        val sc = new SparkContext(sparkConf)
        val rdd = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
        val list = List(("a", 1), ("b", 2), ("c", 3))
        //Broadcast变量只可读
        val bc: Broadcast[List[(String, Int)]] = sc.broadcast(list)
        //模拟join操作，这种方式无需shuffle
        val rdd2 = rdd.map(t => { 
          var temp = 0
          for (ele <- bc.value) { 
            if (ele._1 == t._1) { 
              temp = ele._2
            }
          }
          (t._1, (t._2, temp))
        })
        rdd2.foreach(println) //(a,(1,1)),(b,(2,2)),(c,(3,3))

[20201028200354691.png]: /images/20221120/5318675114c64e8881a63a84a93f4542.png