之前對Broadcast有分析，但是不夠深入《Spark2.3（四十三）：Spark Broadcast總結》，本章對其實現過程以及原理進行分析。
帶着以下幾個問題去寫本篇文章：
1）driver端如何實現broadcast的裝備，是否會把broadcast數據發送給executor端？
2）executor如何獲取到broadcast數據？
導入
Spark一個非常重要的特征就是共享變量。共享變量分為廣播變量（broadcast variable）和累加器（Accumulators）。
- 廣播變量可以在driver程序中寫入，在executor端讀取。
- 累加器在executors中寫入，而在驅動程序(driver端)讀取。
但本章只講解broadcast變量
Spark官網“共享變量”簡介請參考：
http://spark.apache.org/docs/latest/rdd-programming-guide.html#broadcast-variables
通常，當傳遞給Spark算子（比如map或reduce）函數在遠程集群節點上執行時，它在函數中使用的所有變量的單獨副本上工作。這些變量被復制到每台服務器上，對遠程服務器上變量的任何更新都不會傳播回driver程序。通常支持跨Tasks的讀寫共享變量性能比較低。也就是說如果在一個算子函數中使用到了某個外部的變量，那么這個變量的值會被拷貝到每個task中。此時每個task只能操作自己的那份變量副本。如果多個task想要共享某個變量，那么這種方式是做不到的。
然而，Spark確實為兩種常見的使用模型提供了兩種有限類型的共享變量：廣播變量和累加器。
廣播變量的概念
廣播變量提供了一種方法，可以在driver端獲取本地數據，並將它的只讀副本分發給運行有executor（與driver一起構成一個應用程序的executor，不是另外一個應用程序的executor）的節點服務器，而不是為每個Task發送新副本。廣播變量允許程序員在每台計算機上緩存只讀變量，而不是將其副本與Tasks一起發送。例如，它們可以有效地為每個節點提供一個大型輸入數據集的副本。Spark還嘗試使用有效的廣播算法來分發廣播變量，以降低通信成本。Spark將值傳遞給Executor（一次），當在Executor中多次使用廣播變量時，Task可以共享它而不會導致重復的網絡傳輸。
Spark應用程序作業的執行由一系列Stage構成，由分布式“shuffle”操作分隔。Spark自動廣播每個Stage中Tasks所需的公共數據。以這種方式廣播的數據以序列化形式緩存，並在運行每個Task之前進行反序列化。這意味着，僅當跨多個Stage的Tasks需要相同的數據或以反序列化形式緩存數據時，顯式創建廣播變量才有用。
廣播變量是通過調用sparkContext.broadcast（v）從變量v創建的。廣播變量是圍繞v的包裝器，它的值可以通過調用value方法來訪問。使用示例：
scala> val broadcastVar = sc.broadcast(Array(1, 2, 3)) broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0) scala> broadcastVar.value res0: Array[Int] = Array(1, 2, 3)
創建廣播變量后，應在集群上運行的任何函數中使用它，而不是使用值v，這樣V就不會多次發送到節點。此外，對象v在廣播后不應進行修改，以確保所有節點都獲得廣播變量的相同值（例如，如果變量稍后被發送到新節點）。
使用廣播變量的場景示例：
- 大表join小表時，將小表廣播掉，以（broadcast join）方式運行，實現優化。
- 廣播機器學習模型以便能夠對我們的數據進行預測
通過SparkContext#broadcast(v)來創建廣播變量，返回的廣播變量其實就是一個實現了Broadcast接口的包裝類（TorrentBroadcast）對象。在使用時，需要在executor上運行的算子（map/reduce等）內部會引用該broadcast包裝的廣播變量。
另外創建后在driver端修改廣播變量中的值，而不做其他操作時，executor是無法拿到driver端修改后的數據的。如果要修改創建后的廣播變量，請參考《Spark2.3（四十二）：Spark Streaming和Spark Structured Streaming更新broadcast總結（二）》。
廣播替代方案
其實我們可以在閉包中捕獲局部變量，以便將數據從driver程序傳輸到worker，但這種方式是每個Task都發送一個副本變量，相對於broadcast方式比較耗費網絡帶寬和內存空間。
另外我們也可以在Executor端采用單例的方式去加載資源文件，來替代broadcast方案，但是這種方案與broadcast還是有區別：
- broadcast不涉及到當executor多個vcore時，並行task之間同時加載單例資源數據線程安全問題（當然單利中可以通過synchronized同步實現線程安全，首次加載並行線程排隊問題會涉及到效率問題）；
- broadcast比較完美情況下，在executor的節點服務器上只有一個executor從driver端通過remote方式讀取副本數據，該節點服務器上的其他executor從local讀取資源，讀取不到時才會remote讀取資源；但單例方式每個executor都需要加載一次資源；
- 存儲結構不同，broadcast是在spark2.0+內存結構中存儲在Storage Memory這塊內存區上，而單例方式的數據是存放到“用戶自定義數據結構”Other內存區上。
使用時注意事項
1）broadcast的定義必須在Driver端，不能再executor端定義；
2）調用unpersist()，unpersist(boolean blocking),destroy(),distroy(boolean blocking)方法這些方法必須在driver端調用。
3）在Driver端可以修改廣播變量的值，在Executor端無法修改廣播變量的值。
具體操作步驟：
a.在driver端調用unpersist(true)方法；
b.對該broadcast實例對象進行重新賦值。
參考《Spark2.2（三十三）：Spark Streaming和Spark Structured Streaming更新broadcast總結（一）》、《Spark2.3（四十二）：Spark Streaming和Spark Structured Streaming更新broadcast總結（二）》
4）能不能將一個RDD使用廣播變量廣播出去？
不能，因為RDD是不存儲數據的。可以將RDD的結果廣播出去。
5）如果executor端用到了Driver的變量，如果不使用廣播變量在Executor有多少task就有多少Driver端的變量副本。
6）如果Executor端用到了Driver的變量，如果使用廣播變量在每個Executor中只有一份Driver端的變量副本。
7）使用時，算子函數內部引用的是broadcast包裝類對象，而不是broadcast.getValue()的值。否則，毫無意義，每個task都會存在一份副本。
Broadcast的實現
首先我們我們看下broadcast的類結構：
SparkContext類：https://github.com/apache/spark/blob/branch-2.4/core/src/main/scala/org/apache/spark/SparkContext.scala，是broadcast使用的入口函數。
Broadcast接口類：https://github.com/apache/spark/blob/branch-2.4/core/src/main/scala/org/apache/spark/broadcast/Broadcast.scala，是broadcast包裝類的接口類。
BroadcastFactory接口類：https://github.com/apache/spark/blob/branch-2.4/core/src/main/scala/org/apache/spark/broadcast/BroadcastFactory.scala，是broadcast包裝的工廠類接口。
TorrentBroadcast類：https://github.com/apache/spark/blob/branch-2.4/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala，是broadcast包裝類的唯一實現類。
TorrentBroadcastFactory類：https://github.com/apache/spark/blob/branch-2.4/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcastFactory.scala，是broadcast包裝的工廠類接口的唯一實現類。
BroadcastManager類：https://github.com/apache/spark/blob/branch-2.4/core/src/main/scala/org/apache/spark/broadcast/BroadcastManager.scala，broadcast管理類，在SparkEnv中引用，SparkContext#broadcast中調用就是SparkContext#env#broadcastManager的。
BlockManager類：https://github.com/apache/spark/blob/branch-2.4/core/src/main/scala/org/apache/spark/storage/BlockManager.scala，blockManager是env中引用對block的管理類，它會把數據根據存儲級別來存儲數據。如果是memory存儲則調用memoryStore來實現存儲，如果是磁盤存儲則使用diskStore來實現存儲。
MemoryStore類：https://github.com/apache/spark/blob/branch-2.4/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala，內存存儲實現類。讀、寫操作。
DiskStore類結構：https://github.com/apache/spark/blob/branch-2.4/core/src/main/scala/org/apache/spark/storage/DiskStore.scala，磁盤存儲實現類。讀、寫操作。
SparkContext中broadcast(v)函數
用來創建broadcast variable的broadcasat(v)函數定義在SparkContext中，SparkContext中broadcas(v)t代碼如下：
/** * Broadcast a read-only variable to the cluster, returning a * [[org.apache.spark.broadcast.Broadcast]] object for reading it in distributed functions. * The variable will be sent to each cluster only once. * * @param value value to broadcast to the Spark nodes * @return `Broadcast` object, a read-only variable cached on each machine */ def broadcast[T: ClassTag](value: T): Broadcast[T] = { assertNotStopped() require(!classOf[RDD[_]].isAssignableFrom(classTag[T].runtimeClass), "Can not directly broadcast RDDs; instead, call collect() and broadcast the result.") val bc = env.broadcastManager.newBroadcast[T](value, isLocal) val callSite = getCallSite logInfo("Created broadcast " + bc.id + " from " + callSite.shortForm) cleaner.foreach(_.registerBroadcastForCleanup(bc)) bc }
調用了SparkContext#broadcast(v)會創建一個廣播變量，並返回一個org.apache.spark.broadcast.Broadcast對象，這樣就可以分布式函數中來讀取該廣播變量的值。該變量會被發送到各個存在executor的節點上。
Broadcast()函數的實現流程如下：
1）判斷需要廣播的變量是否是分布式變量，若是則會終止函數，報告"Can not directly broadcast RDDs; instead, call collect() and broadcast the result.”的錯誤。
2）通過BroadcastManger#newBroadcast函數來創建廣播變量，並返回一個Broadcast對象，Broadcast只是一個接口類，真正返回的是TorrentBroadcast類的對象。
3）注冊broadcast的cleanup函數，實際上相當於注冊到ContextCleaner中，后邊ContextCleaner內部有一個線程會默認每30分鍾觸發一次對弱引用的broadcast等的清理工作。
4）返回新創建的實現了Broadcast接口的類對象（或者說，返回broadcast包裝類對象）。
BroadcastManager類
BroadcastManager在SparkContext#broadcast（v）中被使用到，因此我們沿着這個路線，來看下這個BroadcastManager的實現類。它是一個broadcast的管理類，用來統一創建broadcast對外的接口：創建，銷毀。
BroadcastManager的類定義如下：
private[spark] class BroadcastManager( val isDriver: Boolean, conf: SparkConf, securityManager: SecurityManager) extends Logging { private var initialized = false private var broadcastFactory: BroadcastFactory = null initialize() // Called by SparkContext or Executor before using Broadcast private def initialize() { synchronized { if (!initialized) { broadcastFactory = new TorrentBroadcastFactory broadcastFactory.initialize(isDriver, conf, securityManager) initialized = true } } } def stop() { broadcastFactory.stop() } private val nextBroadcastId = new AtomicLong(0) private[broadcast] val cachedValues = { new ReferenceMap(AbstractReferenceMap.HARD, AbstractReferenceMap.WEAK) } def newBroadcast[T: ClassTag](value_ : T, isLocal: Boolean): Broadcast[T] = { broadcastFactory.newBroadcast[T](value_, isLocal, nextBroadcastId.getAndIncrement()) } def unbroadcast(id: Long, removeFromDriver: Boolean, blocking: Boolean) { broadcastFactory.unbroadcast(id, removeFromDriver, blocking) } }
該類的構造函數流程如下：
// 是否已經初始 private var initialized = false private var broadcastFactory: BroadcastFactory = null initialize() // 生成廣播變量的id，該id是唯一的，這里先初始化，會在創建broadcast變量時進行自增操作 private val nextBroadcastId = new AtomicLong(0)
1）定義了兩個proviate的變量：initialized，nextBroadcastId。其中nextBroadcastId為每個廣播變量生成一個唯一的id，在創建broadcast變量時，會通過nextBroadcastId,getAndIncrement()進行自增，並調用initialize()函數來進行初始化。
2）initialize()函數的實現邏輯如下：
// Called by SparkContext or Executor before using Broadcast private def initialize() { synchronized { // 加鎖 if (!initialized) { // 初始化broadcastFactory變量，這里創建了TorrentBroadcastFactory對象 broadcastFactory = new TorrentBroadcastFactory // 調用TorrentBroadcastFactory的initialize函數來初始化 broadcastFactory.initialize(isDriver, conf, securityManager) // 把initialized設置為true，同一個對象只初始化一次 initialized = true } } }
2.1）初始化broadcastFactory變量，這里創建了TorrentBroadcastFactory對象；
2.2）調用TorrentBroadcastFactory#initialize()進行初始化，可以查閱TorrentBroadcastFactory，會發現這個initialize()方法實現為空，什么也不做；
2.3）把initialize變量設置為true，並結合synchronized實現broadcastFactory只能被實例化一次。
BroadcastManager中定義了unbroadcast（）方法
實現broadcast清理功能：刪除executor上與此TorrentBroadcast關聯的所有持久化塊。如果removeFromDriver為true，也要刪除driver程序上的這些持久block。
- id: Long, 這里指broadcatId
- removeFromDriver: Boolean, 是否從driver端移除broadcast持久化block數據，否則driver端會殘留broadcast持久化block數據。
- blocking: Boolean，是否同步等待完成才返回。如果false，則會異步的刪除executor上broadcast副本數據。
BroadcastManager中定義了newBroadcast（）方法
該方法是SparkConetxt#broadcast（）中調用用來創建broadcast對象使用的，那么我們看下其具體實現：
def newBroadcast[T: ClassTag](value_ : T, isLocal: Boolean): Broadcast[T] = {
broadcastFactory.newBroadcast[T](value_, isLocal, nextBroadcastId.getAndIncrement())
}
1）根據上邊分析，在BroadcastManager初始化時，broadcastFactory給賦值為TorrentBroadcastFactory，那么這個broadcastFactory其實就是TorrentBroadcastFactory的實例對象；
2）分析到這里，我們可以清楚的知道真正創建broadcast包裝類的實現是在TorrentBroadcastFactory#newBroadcast()實現的，那么接下來我們來查閱TorrentBroadcastFactory的實現。
TorrentBroadcastFactory類
該類實現了一個類似於BitTorrent的協議，通過該協議把廣播數據分發到各個executor中。這些操作其實是在類TorrentBroadcast中實現的。
比特流（BitTorrent）是一種內容分發協議，由布拉姆·科恩自主開發（2003年，軟件工程師Bram Cohen發明了BitTorrent協議）。它采用高效的軟件分發系統和點對點技術共享大體積文件（如一部電影或電視節目），並使每個用戶像網絡重新分配結點那樣提供上傳服務。一般的下載服務器為每一個發出下載請求的用戶提供下載服務，而BitTorrent的工作方式與之不同。分配器或文件的持有者將文件發送給其中一名用戶，再由這名用戶轉發給其它用戶，用戶之間相互轉發自己所擁有的文件部分，直到每個用戶的下載都全部完成。這種方法可以使下載服務器同時處理多個大體積文件的下載請求，而無須占用大量帶寬。
TorrentBroadcastFactory的代碼定義如下：
/** * A [[org.apache.spark.broadcast.Broadcast]] implementation that uses a BitTorrent-like * protocol to do a distributed transfer of the broadcasted data to the executors. Refer to * [[org.apache.spark.broadcast.TorrentBroadcast]] for more details. */ private[spark] class TorrentBroadcastFactory extends BroadcastFactory { override def initialize(isDriver: Boolean, conf: SparkConf, securityMgr: SecurityManager) { } override def newBroadcast[T: ClassTag](value_ : T, isLocal: Boolean, id: Long): Broadcast[T] = { new TorrentBroadcast[T](value_, id) } override def stop() { } /** * Remove all persisted state associated with the torrent broadcast with the given ID. * @param removeFromDriver Whether to remove state from the driver. * @param blocking Whether to block until unbroadcasted */ override def unbroadcast(id: Long, removeFromDriver: Boolean, blocking: Boolean) { TorrentBroadcast.unpersist(id, removeFromDriver, blocking) } }
TorrentBroadcast類
TorrentBroadcasat是真正廣播變量操作實現類
該類實現機制：
1）driver端將序列化對象分成多個小block，並將這些block存儲到driver端的blockManager中，blockManager是上會根據當前存儲類型（MEMORY_AND_DISK_SER，這也是當driver內存不夠時，driver也不內存溢出的原因）去存儲，優先使用內存存儲，調用MemoryStore來存儲，如果內存不夠，則使用DiskStore來進行磁盤存儲。
2）在每個executor上，executor使用到broadcast對象時，會嘗試從當前executor的blockManager中獲取數據，若不存在，則遠程從driver或其他executor(如果可用)中獲取對象block，一旦獲取到block，它就會將block存放到broadcastCache中，為其他executor來獲取數據做好准備。
3）通過這種方式，可以防止driver成為（向每個executor）發送廣播副本的瓶頸。如果driver向每個executor都要發送廣播副本則會導致driver網絡帶寬成為瓶頸，效率也會比較低。
代碼分析：
該類的構造函數：
1）通過readBroadcastBlock函數構造broadcast，並存放到broadcastCache中。在driver端，若需要value值，它會直接從本地的block manager中讀取數據。readBroadcastBlock函數的實現邏輯如下：
readBlock業務：
獲取流程為：嘗試從當前executor的blockManager中獲取數據，若不存在，則遠程從driver或其他executor(如果可用)中獲取對象block，獲取到后存放到broadcastCache中。
1）從當前executor本地獲取數據塊：SparkEnv.get.broadcastManager.cachedValues獲取對應broadcastId的數據塊值：broadcastCache.get(broadcastId)
2）從blockManager中獲取對應id的廣播變量的值：blockManager.getLocalValues(broadcastId)
3）若從blockManager中獲取到了該變量的值，則：broadcastCache.put(broadcastId, x)
4）若不能從blockManager中獲取值，則調用readBlocks函數來讀取數據塊。該函數會從driver或其他的executors中讀取該變量的數據。
2）設置配置信息：setConf(SparkEnv.get.conf)
3）初始化廣播變量的唯一值：private val broadcastId=BraodcastBlockId(id)
4）調用writeBlocks把廣播變量划分成多個塊，並保存到blockManager中。
注意：其中該類中有些屬性被@transient修飾的，被@transient修飾的屬性不能被序列化。
Executor端如何獲取broadcast值
從實際開發中，我們常先定義的broadcast，然后把broadcast對象傳遞給算子（比如：MapFunction）內部調用。傳遞給算子時，是傳遞的broadcast數據的包裝類對象（實際上就是TorrentBroadcast對象），在內部通過broadcast#getValue()獲取broadcast值。舉例說明：
StructType resulStructType = new StructType(); resulStructType = resulStructType.add("int_id", DataTypes.StringType, false); resulStructType = resulStructType.add("job_result", DataTypes.StringType, true); ExpressionEncoder<Row> resultEncoder = RowEncoder.apply(resulStructType); JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext().getConf()); Map<String, String> resource = new java.util.HashMap<String, String>(); for (int i = 0; i < 10000; i++) { resource.put(String.valueOf(i), String.valueOf(i)); } Broadcast<Map<String, String>> broadcastMap = javaSparkContext.broadcast(resource); sourceDataset = sourceDataset.map(new MapFunction<Row, Row>() { private static final long serialVersionUID = 1L; @Override public Row call(Row row) throws Exception { int int_id_idx = row.schema().fieldIndex("int_id"); Object int_idObject = row.get(int_id_idx); String int_id = int_idObject.toString();
Map<String, String> resources = broadcastMap.getValue(); String job_result = resources.get(int_id); Object[] values = new Object[2]; values[0] = int_id; values[1] = job_result; return RowFactory.create(values); } }, resultEncoder); sourceDataset.printSchema();
那么，我們接下來從源碼分析下，這個TorrentBroadcast#getValue()方法是如何實現的：
private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long) extends Broadcast[T](id) with Logging with Serializable { /** * Value of the broadcast object on executors. This is reconstructed by [[readBroadcastBlock]], * which builds this value by reading blocks from the driver and/or other executors. * * On the driver, if the value is required, it is read lazily from the block manager. */ @transient private lazy val _value: T = readBroadcastBlock() /** The compression codec to use, or None if compression is disabled */ @transient private var compressionCodec: Option[CompressionCodec] = _ /** Size of each block. Default value is 4MB. This value is only read by the broadcaster. */ @transient private var blockSize: Int = _ private def setConf(conf: SparkConf) { compressionCodec = if (conf.getBoolean("spark.broadcast.compress", true)) { Some(CompressionCodec.createCodec(conf)) } else { None } // Note: use getSizeAsKb (not bytes) to maintain compatibility if no units are provided blockSize = conf.getSizeAsKb("spark.broadcast.blockSize", "4m").toInt * 1024 checksumEnabled = conf.getBoolean("spark.broadcast.checksum", true) } setConf(SparkEnv.get.conf) private val broadcastId = BroadcastBlockId(id) /** Total number of blocks this broadcast variable contains. */ private val numBlocks: Int = writeBlocks(obj) /** Whether to generate checksum for blocks or not. */ private var checksumEnabled: Boolean = false /** The checksum for all the blocks. */ private var checksums: Array[Int] = _ override protected def getValue() = { _value } 。。。 }
從上邊代碼中我們可以知道：
1）傳遞給executor的算子（類似MapFunction）的broadcast包裝對象就是TorrentBroadcast對象。
2）“Broadcast包裝對象”傳遞給executor時，攜帶的傳遞屬性包含TorrentBroadcast的broadcastId,numBlocks等（除被@transient修飾外的屬性）：
在executor端獲取值是通過“Broadcast包裝對象”#getValue()方法獲取值得，driver端傳遞過去的只是TorrentBroadcast對象的序列化字符串（被task引用的，因此隨task而傳遞[這點可質疑]，是不是broadcast有特殊機制對task引用的broadcast對象采用單獨傳遞方式？），當executor端反序列化task時，會把TorrentBroadcast反序列化為對象，通過調用broadcast#getValue（）方法獲取broadcast的值。
TorrentBroadcast在driver端被序列化時，只能對TorrentBroadcast的非@transient 修飾的屬性進行序列化（被@transient修飾的屬性不能被序列化），因為TorrentBroadcast的_value是被@transient修飾，所以broadcast的值在driver端並不能被反序列化傳遞給executor。
這也是實現driver端傳遞broadcast給executor時，只傳遞broadcastId,numBlocks等屬性（除去_value值外）的真正實現。
3）在executor端，執行map算子時，會調用TorrentBroadcast對象的getVlaue()方法，此時實際上是讀取_value=readBroadcastBlock()，也就是嘗試從當前executor的blockManager讀取，當前executor中讀取不到則從當前節點其他executor中獲取，如果還是獲取不到則從遠程driver讀取。
斷點監控結果截圖：
參考
《spark2原理分析-廣播變量(Broadcast Variables)的實現原理》
《SparkSQL中的三種Join及其具體實現（broadcast join、shuffle hash join和sort merge join）》