Spark的TorrentBroadcast：实现

依据Spark 1.4版

序列化和反序列化

前边提到，TorrentBroadcast的关键就在于特殊的序列化和反序列化设置。1.1版的TorrentBroadcast实现了自己的readObject和writeObject方法，但是1.4.1版的TorrentBroadcast没有实现自己的readObject方法，那么它是如何进行序列化和反序列化的呢？

// obj就是被广播的对象
 private val numBlocks: Int = writeBlocks(obj)

  override protected def getValue() = {
    _value
  }

@transient private lazy val _value: T = readBroadcastBlock()

可以认为TorrentBroadcast对象经过了三个主要阶段的处理：构造器，序列化，反序列化

构造器

在构造TorrentBroadcast对象时，numBlocks会被初始化，此时writeBlocks会被执行。writeBlocks会执行把obj序列化，分块，存储进BlockManager等操作。

而_value域是lazy的，因此在TorrentBroadcast对象初始化时，_value不会初始化，readBroadcastBlock也不会执行。

序列化

当在driver端对RDD调用一个action时，会生成Task对象，Task对象引用到的对象会被序列化，然后对每一个task，反序列化一个Task对象。

TorrentBroadcast需要保证被广播的对象不会随Task一起序列化。需要注意以下两点：

private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
  extends Broadcast[T](id) with Logging with Serializable {
　　……
}

@transient private lazy val _value: T = readBroadcastBlock()

Scala的构造函数里的参数并不一定会成为对象的字段，像obj这种只是用来构造对象、没有被用于实现方法的构造器参数，不会成为TorrentBroadcast的字段，因此不会被序列化。

而_value尽管引用了被广播的数据，但它是@transient的，因此也不会被序列化。

反序列化

反序列化的关键在于，_value不会被反序列化。因此，如果某个executor没有task使用TorrentBroadcast的value方法，被广播的数据就不会被在这个executor端获取。

实现这种功能的关键在于Scala的lazy val。

首先，考虑这个问题：lazy val可能被多个线程同时访问，这会触发lazy val的初始化，但是需要保证这个初始化的过程就线程安全的，即lazy val只被初始化一次，且初始化的结果对所有线程可见。实现这种行为，最简单的做为是使用this做同步，但是这样的效率会很低，而Scala实现lazy val使用了一种效率更高的方法，但不管怎么做，lazy val比普通的val的访问效率会降低。

举一个Double-checked locking idiom, sweet in Scala!中的例子：

lazy val myLazyField = create();

会被编译成：

   public volatile int bitmap$0;
   private Object myLazyField;

   public String myLazyField() {
        if((bitmap$0 & 1) == 0)
        {
            synchronized(this)
            {
                if((bitmap$0 & 1) == 0)
                {
                    myLazyField = ...
                    bitmap$0 = bitmap$0 | 1;
                }
            }
        }
        return myLazyField;
    }

即通过一个volatile变量来判断这个lazy val是否已经初始化，通过双重检查加锁来做初始化。

现在有了新的问题：

1. 默认的序列化过程是否会触发lazy val被初始化呢？

2. 如果在TorrentBroadcast对象被序列化之前，lazy val被访问，触发了初始化过程，那么被广播的数据相关于作为TorrentBroadcast的一个field，也会被序列化。

问题1的答案是不会触发。问题2的答案_value需要被注明是transient，就像TorrentBroadcast里所做的一样。

所以，在函数中如果经常使用Broadcast.value方法返回的对象时，比如在循环中使用它，最后先在循环外创建一个对这个对象的引用，以减少一些开销。

但是，lazy val的这种线程安全机制对于TorrentBroadcast是浪费的。因为Broadcast变量是随Task一起序列化的，每个线程有自己的Task对象，也就是线程间不共享Broadcast对象。实际上，为了保证同一个JVM上运行的不同task得到同样的被广播的对象，readBroadcastBlock方法是使用TorrentBroadcast这个class做了同步，

下面来看一下把被广播的对象分块存储的过程

将广播的对象分块存储

这一步是在TorrentBroadcast对象初始化时候做的。

由 

val numBlocks: Int = writeBlocks(obj)

触发。下面看一下writeBlocks方法

writeBlocks

  private def writeBlocks(value: T): Int = {
    // Store a copy of the broadcast variable in the driver so that tasks run on the driver
    // do not create a duplicate copy of the broadcast variable's value.
    SparkEnv.get.blockManager.putSingle(broadcastId, value, StorageLevel.MEMORY_AND_DISK,
      tellMaster = false)
    val blocks =
      TorrentBroadcast.blockifyObject(value, blockSize, SparkEnv.get.serializer, compressionCodec)
    //blocks的类型是Array[ByteBuffer]
    blocks.zipWithIndex.foreach { case (block, i) =>
      SparkEnv.get.blockManager.putBytes(
        BroadcastBlockId(id, "piece" + i),//以BroadcastBlockId为BlockId存储
        block,
        StorageLevel.MEMORY_AND_DISK_SER,
        tellMaster = true)
    }
    blocks.length
  }

正如代码中的注释所说的，writeBlocks会首先把被广播的对象用putSingle方法放在driver的BlockManager里，这是为了当在driver运行task时，不会额外创建一个被广播的对象的副本。若没有这一步，在driver端运行task时，会和executor端一样，通过Broadcast对象的value方法新建一个被广播的对象，这就使得driver端有两份这个对象。但实际上driver端运行task的情况并不常见。所以这里最好根据conf判断下是否有必要这么做。

接下来，使用伴生对象的blockifyObject方法把对象分块，得到的结果是一个ByteBuffer的数组。然后把这些块存进BlockManager, 这里有两点需要注意：

1. 把块存进BlockManager时，使用的id是BroadcastBlockId(id, "piece" + i)。也就是说跟据Broadcast对象的id，以及总共的块的数量就可以还原出所有的块存储时所使用的id。这也就是为什么TorrentBroadcast要有numBlocks这个field的原因。而id字段是Broadcast这个虚类里的val， 所以根据TorrentBroadcast对象的字段，即可以它所划分的所有block的id。在从这些块还原被broadcast的对象时，也的确是这么做的。

2. 把划分出的块存储进BlockManager时，tellMaster字段的值为true，这就使得master可以知道哪个BlockManager存储了这个块，因此executor端的BlockManager最初的时候才能从driver端的BlockManager获取这个块。相反的是，writeBlocks第一句putSingle时，tellMaster是false，因为并不准备让其它BlockManager获取putSingle进去的对象。

blockifyObject

blockifyObject作的工作就是将被广播的对象序列化，如果启用了压缩就进行压缩，然后将得到的字节流写入到一系列字节数组中。

它的返回值类型为：Array[ByteBuffer], 之所有是ByteBuffer, 是为了BlockManager使用方便，因为BlockManager的putBytes方法接受ByteBuffer作为参数。

  def blockifyObject[T: ClassTag](
      obj: T,
      blockSize: Int,
      serializer: Serializer,
      compressionCodec: Option[CompressionCodec]): Array[ByteBuffer] = {
    val bos = new ByteArrayChunkOutputStream(blockSize)
    val out: OutputStream = compressionCodec.map(c => c.compressedOutputStream(bos)).getOrElse(bos)
    val ser = serializer.newInstance()
    val serOut = ser.serializeStream(out)
    serOut.writeObject[T](obj).close()
    bos.toArrays.map(ByteBuffer.wrap)
  }

它实现的关键在于ByteArrayChunkOutputStream, 这个类实现了Java的OutputStream接口。它的主体部分如下：

private[spark] class ByteArrayChunkOutputStream(chunkSize: Int) extends OutputStream {

  private val chunks = new ArrayBuffer[Array[Byte]]
  private var lastChunkIndex = -1
  private var position = chunkSize
  override def write(b: Int): Unit = {
    allocateNewChunkIfNeeded()
    chunks(lastChunkIndex)(position) = b.toByte
    position += 1
  }

　override def write(bytes: Array[Byte], off: Int, len: Int): Unit = { ... }

　def toArrays: Array[Array[Byte]] = { ... }

 ... 

}

即，它在内部使用一些长度等于chunkSize的数组来存储被写入的字节。

组装还原被广播的对象

在executor端(如果有task在driver执行的话，也可以是在driver端)需要把被切块后的对象组装起来，还原成被广播的对象。这是通过对lazy val _value访问触发的。

 @transient private lazy val _value: T = readBroadcastBlock()

readBroadcast会首先在本地的BlockManager寻找之前存入的被广播的对象，因此如果同一个executor中已经有task访问过_value，那么它就能直接取到已被放入本地BlockManager中的对象，

如果本地还没有, 那么就会调用readBlocks获取组成这个对象的块，然后用unblockifyObject还原这个对象，接着把它放入BlockManager，以使得同一个executor的其它task不必重复组装还原。

 private def readBroadcastBlock(): T = Utils.tryOrIOException {
    TorrentBroadcast.synchronized {
      setConf(SparkEnv.get.conf)
      //从本地的blockManager里读这个被broadcast的对象，根据broadcastId
      SparkEnv.get.blockManager.getLocal(broadcastId).map(_.data.next()) match {
        case Some(x) => //本地有
          x.asInstanceOf[T]

        case None => //本地无
          logInfo("Started reading broadcast variable " + id)
          val startTimeMs = System.currentTimeMillis()
          val blocks = readBlocks()//如果本地没有broadcastId对应的broadcast的block，就读
          logInfo("Reading broadcast variable " + id + " took" + Utils.getUsedTimeMs(startTimeMs))

          val obj = TorrentBroadcast.unBlockifyObject[T](
            blocks, SparkEnv.get.serializer, compressionCodec)
          // Store the merged copy in BlockManager so other tasks on this executor don't
          // need to re-fetch it.
          SparkEnv.get.blockManager.putSingle( //读了之后再放进BlockManager
            broadcastId, obj, StorageLevel.MEMORY_AND_DISK, tellMaster = false)
          obj
      }
    }
  }

这里有一个细节是，组装还原之后的对象被用putSingle放入BlockManager, 存储级别为MEMORY_AND_DISK，这就意味着，在MemoryStore无法容纳被广播的对象时，同一个executor的两个task可能会获取两个不同的对象(需要研究下BlockManager相关的代码才能确定)。如果这种情况发生，而被广播的对象是线程安全的，那么就是对内存的浪费。如果这种情况不发生，一个executor的所有task共享一个被广播的对象，那么可能会产生线程安全的问题。但是无论如何，使用被广播的对象时，需要以只读的方式，对它的修改可能会产生问题。

TorrentBroadcast是通过readBlocks获取构成序列化后的对象的块。

  /** Fetch torrent blocks from the driver and/or other executors. */
  private def readBlocks(): Array[ByteBuffer] = {
    //获取到的block被存在本地的BlockManager中并且上报给driver,这样其它的executor就可以从这个executor获取这些block了
    val blocks = new Array[ByteBuffer](numBlocks)
    val bm = SparkEnv.get.blockManager

    //需要shuffle，避免所有executor以同样的顺序下载block，使得driver依然是瓶颈
    for (pid <- Random.shuffle(Seq.range(0, numBlocks))) {
      val pieceId = BroadcastBlockId(id, "piece" + pid)//组装BroadcastBlockId
      logDebug(s"Reading piece $pieceId of $broadcastId")
      // 先试着从本地获取，因为之前的尝试可能已经获取了一些block
      def getLocal: Option[ByteBuffer] = bm.getLocalBytes(pieceId)
      def getRemote: Option[ByteBuffer] = bm.getRemoteBytes(pieceId).map { block =>
        //如果从remote获取了block，就把它存在本地的BlockManager
        SparkEnv.get.blockManager.putBytes(
          pieceId,
          block,
          StorageLevel.MEMORY_AND_DISK_SER,
          tellMaster = true)
        block
      }
      val block: ByteBuffer = getLocal.orElse(getRemote).getOrElse(
        throw new SparkException(s"Failed to get $pieceId of $broadcastId"))
      blocks(pid) = block
    }
    blocks
  }

readBlocks还是很简单易懂的，只是这里使用putBytes时，使用的存储级别是MEMORY_AND_DISK_SER，有些奇怪，不知道为啥对于这些bytes还需要序列化。

总结

TorrentBroadcast的实现有一些巧妙的细节，但是整体的代码还是很简洁，也比较容易理解。之所以有如此少的代码，是因为BlockManager已经提供了足够的基础设施。