A broadcast variable. Broadcast variables allow the programmer to keep a read-only variable cached on each machine rather than shipping a copy of it with tasks. 
They can be used, for example, to give every node a copy of a large input dataset in an efficient manner. Spark also attempts to distribute broadcast variables using efficient broadcast algorithms to reduce communication cost. Broadcast variables are created from a variable v by calling org.apache.spark.SparkContext.broadcast. The broadcast variable is a wrapper around v, and its value can be accessed by calling the value method. 
The interpreter session below shows this:

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)

After the broadcast variable is created, it should be used instead of the value v in any functions run on the cluster so that v is not shipped to the nodes more than once. In addition, the object v should not be modified after it is broadcast in order to ensure that all nodes get the same value of the broadcast variable (e.g. if the variable is shipped to a new node later).

A BitTorrent-like implementation of org.apache.spark.broadcast.Broadcast. 
The mechanism is as follows: 
The driver divides the serialized object into small chunks and stores those chunks in the BlockManager of the driver. 
On each executor, the executor first attempts to fetch the object from its BlockManager. 
If it does not exist, it then uses remote fetches to fetch the small chunks from the driver and/or other executors if available. 
Once it gets the chunks, it puts the chunks in its own BlockManager, ready for other executors to fetch from. 
This prevents the driver from being the bottleneck in sending out multiple copies of the broadcast data (one per executor). 
When initialized, TorrentBroadcast objects read SparkEnv.get.conf.

 private def readBroadcastBlock(): T = Utils.tryOrIOException {

   TorrentBroadcast.synchronized {

     val broadcastCache = SparkEnv.get.broadcastManager.cachedValues

     Option(broadcastCache.get(broadcastId)).map(_.asInstanceOf[T]).getOrElse {

       setConf(SparkEnv.get.conf)

       val blockManager = SparkEnv.get.blockManager

       blockManager.getLocalValues(broadcastId) match {

         case Some(blockResult) =>

           if (blockResult.data.hasNext) {

             val x = blockResult.data.next().asInstanceOf[T]

             releaseLock(broadcastId)

             if (x != null) {

               broadcastCache.put(broadcastId, x)

             }

             x

           } else {

             throw new SparkException(s"Failed to get locally stored broadcast data: $broadcastId")

           }

         case None =>

           logInfo("Started reading broadcast variable " + id)

           val startTimeMs = System.currentTimeMillis()

           val blocks = readBlocks()

           logInfo("Reading broadcast variable " + id + " took" + Utils.getUsedTimeMs(startTimeMs))

           try {

             val obj = TorrentBroadcast.unBlockifyObject[T](

               blocks.map(_.toInputStream()), SparkEnv.get.serializer, compressionCodec)

             // Store the merged copy in BlockManager so other tasks on this executor don't

             // need to re-fetch it.

             val storageLevel = StorageLevel.MEMORY_AND_DISK

             if (!blockManager.putSingle(broadcastId, obj, storageLevel, tellMaster = false)) {

               throw new SparkException(s"Failed to store $broadcastId in BlockManager")

             }

             if (obj != null) {

               broadcastCache.put(broadcastId, obj)

             }

             obj

           } finally {

             blocks.foreach(_.dispose())

           }

       }

     }

   }

 }

 /** Fetch torrent blocks from the driver and/or other executors. */

 private def readBlocks(): Array[BlockData] = {

   // Fetch chunks of data. Note that all these chunks are stored in the BlockManager and reported

   // to the driver, so other executors can pull these chunks from this executor as well.

   val blocks = new Array[BlockData](numBlocks)

   val bm = SparkEnv.get.blockManager

   for (pid <- Random.shuffle(Seq.range(0, numBlocks))) {

     val pieceId = BroadcastBlockId(id, "piece" + pid)

     logDebug(s"Reading piece $pieceId of $broadcastId")

     // First try getLocalBytes because there is a chance that previous attempts to fetch the

     // broadcast blocks have already fetched some of the blocks. In that case, some blocks

     // would be available locally (on this executor).

     bm.getLocalBytes(pieceId) match {

       case Some(block) =>

         blocks(pid) = block

         releaseLock(pieceId)

       case None =>

         bm.getRemoteBytes(pieceId) match {

           case Some(b) =>

             if (checksumEnabled) {

               val sum = calcChecksum(b.chunks(0))

               if (sum != checksums(pid)) {

                 throw new SparkException(s"corrupt remote block $pieceId of $broadcastId:" +

                   s" $sum != ${checksums(pid)}")

               }

             }

             // We found the block from remote executors/driver's BlockManager, so put the block

             // in this executor's BlockManager.

             if (!bm.putBytes(pieceId, b, StorageLevel.MEMORY_AND_DISK_SER, tellMaster = true)) {

               throw new SparkException(

                 s"Failed to store $pieceId of $broadcastId in local BlockManager")

             }

             blocks(pid) = new ByteBufferBlockData(b, true)

           case None =>

             throw new SparkException(s"Failed to get $pieceId of $broadcastId")

         }

     }

   }

   blocks

 }