Spark - 廣播變量 & 累加器

廣播變量 broadcast variable

1、廣播變量的意義

如果我們要在分布式計算裡面分發大對象，例如：字典，集合，黑白名單等，這個都會由Driver端進行分發，一般來講，如果這個變量不是廣播變量，那麼每個task就會分發一份，這在task數目十分多的情況下Driver的帶寬會成為系統的瓶頸，而且會大量消耗task伺服器上的資源，如果将這個變量聲明為廣播變量，那麼隻是每個executor擁有一份，這個executor啟動的task會共享這個變量，節省了通信的成本和伺服器的資源。

如：一個spark應用有50個executor，1000個tasks，一個10M大小的資料，不使用廣播變量，則需要10M*1000 = 10G的記憶體，而使用廣播變量則隻需要10M * 50 = 500M記憶體

2、代碼樣例

package com.test.bigdata

import org.apache.spark.{SparkConf, SparkContext}

object BroadcastApp {

  def main(args: Array[String]) {
    val sparkConf = new SparkConf()
      .setMaster("local[2]").setAppName("SparkContextApp")

    val sc = new SparkContext(sparkConf)

    //    commonJoin(sc)

    broadcastJoin(sc)
    sc.stop()
  }

  def broadcastJoin(sc: SparkContext): Unit = {
    // 假設 a1是個小表
    val a1 = sc.parallelize(Array(("1", "大米"), ("2", "洋芋"), ("29", "小花"))).collectAsMap()
  	//廣播
    val a1Broadcast = sc.broadcast(a1) 

    sc.longAccumulator("").add(1)

    val f11 = sc.parallelize(Array(("29", "深圳", 18), ("10", "北京", 2)))
      .map(x => (x._1, x))

    f11.mapPartitions(partition => {
   	 // 擷取廣播裡面的内容
      val a1Stus = a1Broadcast.value 
      for ((key, value) <- partition if (a1Stus.contains(key)))
        yield (key, a1Stus.getOrElse(key,""), value._2, value._3)
    })
  }

  def commonJoin(sc: SparkContext): Unit = {

    // a1 join f11 on a1.id = f11.id   ==> 29,"小花","深圳",18
    val a1 = sc.parallelize(Array(("1", "大米"), ("2", "洋芋"), ("29", "小花"))).map(x => (x._1, x))

    val f11 = sc.parallelize(Array(("29", "深圳", 18), ("10", "北京", 2))).map(x => (x._1, x))

    a1.join(f11).map(x => {
      x._1 + " , " + x._2._1._2 + " , " + x._2._2._2 + " , " + x._2._2._3
    }).collect()


  }
}
           

3、注意事項

• 廣播變量不能過大
• 廣播變量是隻讀屬性，不能修改，在Driver端可以修改廣播變量的值，重新進行廣播，在Executor端無法修改廣播變量的值。

• 必須将RDD進行action操作之後在進行廣播

  val a1 = sc.parallelize(Array((“1”, “大米”), (“2”, “洋芋”), (“29”, “小花”))).collectAsMap()

  val a1Broadcast = sc.broadcast(a1) //廣播

累加器 accumulator

1、累加器

在spark應用程式中，我們經常會有這樣的需求，如異常監控，調試，記錄符合某特性的資料的數目，這種需求都需要用到計數器，如果一個變量不被聲明為一個累加器，那麼它将在被改變時不會再driver端進行全局彙總，即在分布式運作時每個task運作的隻是原始變量的一個副本，并不能改變原始變量的值，但是當這個變量被聲明為累加器後，該變量就會有分布式計數的功能。

2、累加器的使用

val conf = new SparkConf()
conf.setMaster("local").setAppName("accumulator")
val sc = new SparkContext(conf)
//定義累加器
val accumulator = sc.accumulator(0) 
//分布式累加
sc.textFile("./words.txt").foreach { x =>{accumulator.add(1)}} 
//擷取累加器的結果
println(accumulator.value) 
sc.stop()