Spark-尚硅谷5-3数据结构：累加器+广播变量

Bertha 。 2022-11-17 05:30 169阅读 0赞

**Spark 计算框架为了能够进行高并发和高吞吐的数据处理，封装了三大数据结构，用于处理不同的应用场景。  
三大数据结构分别是：**  
![在这里插入图片描述][20210410105145340.png]

# 5.2 累加器 #

## 5.2.1 实现原理 ##

![在这里插入图片描述][20210410101808219.png]

## 5.2.2 基础编程 ##

## 5.2.2.1 系统累加器 ##

**不起作用的方法**

package com.atguigu.bigdata.spark.core.acc
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark01_Acc { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
            val sc = new SparkContext(sparConf)
    
            val rdd = sc.makeRDD(List(1,2,3,4))
    
            // reduce : 分区内计算，分区间计算
            //val i: Int = rdd.reduce(_+_)
            //println(i)
            var sum = 0
            rdd.foreach(
                num => { 
                    sum += num
                }
            )
            println("sum = " + sum)
    
            sc.stop()
    
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]

**Spark提供的累加器**

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

package com.atguigu.bigdata.spark.core.acc
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark02_Acc { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
            val sc = new SparkContext(sparConf)
    
            val rdd = sc.makeRDD(List(1,2,3,4))
    
            // 获取系统累加器
            // Spark默认就提供了简单数据聚合的累加器
            val sumAcc = sc.longAccumulator("sum")
    
            //sc.doubleAccumulator
            //sc.collectionAccumulator
    //行动算子
            rdd.foreach(
                num => { 
                    // 使用累加器
                    sumAcc.add(num)
                }
            )
    
            // 获取累加器的值
            println(sumAcc.value)
    
            sc.stop()
    
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]

**使用累加器的注意点：**

// 少加（不计算）：转换算子中调用累加器，如果没有行动算子的话，那么不会执行
            // 多加（计算多次）：累加器全局共享，调用几次行动算子，转换算子中的累加器就会执行几次
            // 一般情况下，累加器会放置在行动算子进行操作

package com.atguigu.bigdata.spark.core.acc
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark03_Acc { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
            val sc = new SparkContext(sparConf)
    
            val rdd = sc.makeRDD(List(1,2,3,4))
    
            // 获取系统累加器
            // Spark默认就提供了简单数据聚合的累加器
            val sumAcc = sc.longAccumulator("sum")
    
            //sc.doubleAccumulator
            //sc.collectionAccumulator
    
            val mapRDD = rdd.map(
                num => { 
                    // 使用累加器
                    sumAcc.add(num)
                    num
                }
            )
            // 获取累加器的值
            // 少加：转换算子中调用累加器，如果没有行动算子的话，那么不会执行
            // 多加：累加器全局共享，调用几次行动算子，转换算子中的累加器就会执行几次
            // 一般情况下，累加器会放置在行动算子进行操作
            mapRDD.collect()
            mapRDD.collect()
            println(sumAcc.value)
    
            sc.stop()
    
        }
    }

## 5.2.2.2 自定义累加器 ##

可以使用累加器避免计算中的shfllec操作  
简单的数据累加都可以用累加器，例如  
word count

package com.atguigu.bigdata.spark.core.acc
    
    import org.apache.spark.util.AccumulatorV2
    import org.apache.spark.{ SparkConf, SparkContext}
    
    import scala.collection.mutable
    
    object Spark04_Acc_WordCount { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
            val sc = new SparkContext(sparConf)
    
            val rdd = sc.makeRDD(List("hello", "spark", "hello"))
    
            // 累加器 : WordCount
            // 创建累加器对象
            val wcAcc = new MyAccumulator()
            // 向Spark进行注册
            sc.register(wcAcc, "wordCountAcc")
    
            rdd.foreach(
                word => { 
                    // 数据的累加（使用累加器）
                    wcAcc.add(word)
                }
            )
    
            // 获取累加器累加的结果
            println(wcAcc.value)
    
            sc.stop()
    
        }
        /* 自定义数据累加器：WordCount 1. 继承AccumulatorV2, 定义泛型 IN : 累加器输入的数据类型 String OUT : 累加器返回的数据类型 mutable.Map[String, Long] 2. 重写方法（6） */
        class MyAccumulator extends AccumulatorV2[String, mutable.Map[String, Long]] { 
    
            private var wcMap = mutable.Map[String, Long]()
    
            // 判断是否初始状态
            override def isZero: Boolean = { 
                wcMap.isEmpty
            }
    
            override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = { 
                new MyAccumulator()
            }
    
            override def reset(): Unit = { 
                wcMap.clear()
            }
    
            // 获取累加器需要计算的值，每个Executor计算累加
            override def add(word: String): Unit = { 
                val newCnt = wcMap.getOrElse(word, 0L) + 1
                wcMap.update(word, newCnt)
            }
    
            // Driver合并多个累加器，多个map的合并
            override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = { 
    
                val map1 = this.wcMap
                val map2 = other.value
    
                map2.foreach{ 
                    case ( word, count ) => { 
                        val newCount = map1.getOrElse(word, 0L) + count
                        map1.update(word, newCount)
                    }
                }
            }
    
            // 返回累加器最终merge的结果
            override def value: mutable.Map[String, Long] = { 
                wcMap
            }
        }
    }

使用

# 5.3 广播变量 #

**分布式共享的只读变量**

**直接使用map方法代替join**

package com.atguigu.bigdata.spark.core.acc
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{ SparkConf, SparkContext}
    
    import scala.collection.mutable
    
    object Spark05_Bc { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
            val sc = new SparkContext(sparConf)
    
            val rdd1 = sc.makeRDD(List(
                ("a", 1),("b", 2),("c", 3)
            ))
    // val rdd2 = sc.makeRDD(List(
    // ("a", 4),("b", 5),("c", 6)
    // ))
            val map = mutable.Map(("a", 4),("b", 5),("c", 6))
    
            // join会导致数据量几何增长，并且会影响shuffle的性能，不推荐使用
            //val joinRDD: RDD[(String, (Int, Int))] = rdd1.join(rdd2)
            //joinRDD.collect().foreach(println)
            // (a, 1), (b, 2), (c, 3)
            // (a, (1,4)),(b, (2,5)),(c, (3,6))
            rdd1.map { 
                case (w, c) => { 
                    val b: Int = map.getOrElse(w, 0)
                    (w, (c, b))
                }
            }.collect().foreach(println)
            sc.stop()
        }
    }

(a,(1,4))
    (b,(2,5))
    (c,(3,6))

缺点是在只使用一个Executor的时候可能操作内存中数据冗余

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 2]

## 5.3.1 实现原理 ##

![在这里插入图片描述][20210410105043152.png]  
多个Task共享这个map 变量，前提是不能改，是能读，**广播变量是分布式共享的只读变量**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 3]

package com.atguigu.bigdata.spark.core.acc
    
    import org.apache.spark.broadcast.Broadcast
    import org.apache.spark.{ SparkConf, SparkContext}
    
    import scala.collection.mutable
    
    object Spark06_Bc { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
            val sc = new SparkContext(sparConf)
    
            val rdd1 = sc.makeRDD(List(
                ("a", 1),("b", 2),("c", 3)
            ))
            val map = mutable.Map(("a", 4),("b", 5),("c", 6))
    
            // 封装广播变量
            val bc: Broadcast[mutable.Map[String, Int]] = sc.broadcast(map)
    
            rdd1.map { 
                case (w, c) => { 
                    // 方法广播变量
                    val b: Int = bc.value.getOrElse(w, 0)
                    (w, (c, b))
                }
            }.collect().foreach(println)
            
            sc.stop()
    
        }
    }

(a,(1,4))
    (b,(2,5))
    (c,(3,6))

[20210410105145340.png]: /images/20221022/f4796d2b89f6407a914ed5a854ab66cd.png
[20210410101808219.png]: /images/20221022/45190a6316df427ba2c07b23e52d67ec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]: /images/20221022/a783778e66884282bf524d447111db82.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]: /images/20221022/72f12341f73846e5b3298b99bc990c63.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 2]: /images/20221022/92901b6d13784062bc1462cefbd2cc4c.png
[20210410105043152.png]: /images/20221022/766eb9d1953c472b8eee121bfd37f800.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 3]: /images/20221022/595cf55c07aa4fc097580f7bb5a1079d.png