Spark的standalone源码分析（四）

承接上文，继续分析sparkcontext初始化中开启的broadcast服务，文中部分内容参考论文“Performance and Scalability of Broadcast in Spark ”；

2.2 BroadcastManager

相比hadoop，spark的优势在于迭代计算，尤其是一些机器学习算法的实现；在这类计算中，经常需要同步large read-only数据，比如字典表、通用的配置文件等。如果将这些数据和运算闭包绑定在一次，那每一次迭代都需要考虑这些数据，没有必要；类似于hadoop的distributeCache，spark通过broadcast服务，将数据同步到每一个worker节点，同时这些数据且只同步一次；另外，论文详细分析了spark默认的broadcast策略(基于HDFS)的性能，并与Chained Streaming
Broadcast，BitTorrent Broadcast和SplitStreaming Broadcast做了比较，有兴趣的朋友可以去看一下。本节会逐一介绍spark源码中httpbroadcast、treebroadcast、bitTorrent broadcast的实现；

首先说一下broadcast的初始化和调用方法；

1) broadcastManager的初始化及调用

broadcastmanager初始化的时候，带有isMaster参数，即master节点和work节点的初始化操作是有区别的；

class BroadcastManager(val isMaster_ : Boolean) extends Logging with Serializable {

private var initialized = false
private var broadcastFactory: BroadcastFactory = null

broadcast的调用，是通过sparkcontext实例调用broadcast方法实现，比如HadoopRDD中通过broadcast同步JobConf对象，sc即sparkcontext的一个实例。

val confBroadcast = sc.broadcast(new SerializableWritable(conf))

而sc.broadcast会初始化一个Broadcast的实体类：

/**
* Broadcast a read-only variable to the cluster, returning a [[spark.Broadcast]] object for
* reading it in distributed functions. The variable will be sent to each cluster only once.
*/
def broadcast[T](value: T) = env.broadcastManager.newBroadcast[T] (value, isLocal)

2）httpbroadcast

spark.broadcast.httpbroadcast是通过在master节点建立http-server(包的是jetty server)的方式来广播对象的。首先，主节点初始化的时候，开启服务，而worker节点从环境变量中获取主节点的serveruri；

def initialize(isMaster: Boolean) {
synchronized {
if (!initialized) {
bufferSize = System.getProperty("spark.buffer.size", "65536").toInt
compress = System.getProperty("spark.broadcast.compress", "true").toBoolean
if (isMaster) {
createServer()
}
serverUri = System.getProperty("spark.httpBroadcast.uri")
initialized = true
}
}
}

sc.broadcast方法广播变量的时候（会在主节点执行），会首先生成一个broadcast_id，然后将<id,value>交由blockMangager管理（注：blockManager并不会把这个blockid的状态report给master节点，这里面blockManager只是起到了cache的作用，确保broadcast广播的变量只同步一次）；由于value声明为transient，所以在broadcast对象序列化的时候会忽略value变量，这样避免大的数据对象在集群中的拷贝；httpbroadcast在write的时候，会将value写入主节点的临时目录下的以broadcast-id命名的文件内。

private[spark] class HttpBroadcast[T](@transient var value_ : T, isLocal: Boolean, id: Long)
extends Broadcast[T](id) with Logging with Serializable {

def value = value_

def blockId: String = "broadcast_" + id

HttpBroadcast.synchronized {
SparkEnv.get.blockManager.putSingle(blockId, value_, StorageLevel.MEMORY_AND_DISK, false)
}

if (!isLocal) {
HttpBroadcast.write(id, value_)
}

def write(id: Long, value: Any) {
val file = new File(broadcastDir, "broadcast-" + id)
val out: OutputStream = if (compress) {
new LZFOutputStream(new FileOutputStream(file)) // Does its own buffering
} else {
new FastBufferedOutputStream(new FileOutputStream(file), bufferSize)
}
val ser = SparkEnv.get.serializer.newInstance()
val serOut = ser.serializeStream(out)
serOut.writeObject(value)
serOut.close()
}

worker节点在反序列化broadcast对象实例的时候，会开始从master节点同步broadcast_id对应的内容；具体实现是，每一个broadcast的派生类都overwrite了一个readObject的方法，该方法会定义从master节点同步数据的策略。在httpbroadcast中，worker节点反序列化broadcast对象获得blockid，首先请求blockmanager，检测blockmanager中是否已经存有该blockid，如果有的的话，则通过blockManager将value数据同步到worker节点，反之，则通过socket将blockid文件的内容同步到work节点，然后将<id,value>存入blockmanager。

// Called by JVM when deserializing an object
private def readObject(in: ObjectInputStream) {
in.defaultReadObject()
HttpBroadcast.synchronized {
SparkEnv.get.blockManager.getSingle(blockId) match {
case Some(x) => value_ = x.asInstanceOf[T]
case None => {
logInfo("Started reading broadcast variable " + id)
val start = System.nanoTime
value_ = HttpBroadcast.read[T](id)
SparkEnv.get.blockManager.putSingle(blockId, value_, StorageLevel.MEMORY_AND_DISK, false)
val time = (System.nanoTime - start) / 1e9
logInfo("Reading broadcast variable " + id + " took " + time + " s")
}
}
}
}

可见，http-broadcast是采用中心节点同步的方式，随着集群规模的扩大，这样的同步方式显然效率不高；

3) TreeBroadcast

broadcast factory初始化treebroadcast时，在master节点，会初始化TrackMultipleValues服务；TrackMultipleValues服务监听MasterTrackerPort端口，处理接受到得三类数据：1)register-broadcast：将发送节点的hostaddress和监听端口信息加入valueToGuideMap；2）unregister-broadcast：将发送节点信息在valueToGuideMap里面置空；3）find-broadcast：根据发送节点的请求的id，返回对应的SourceInfo；

def initialize(isMaster__ : Boolean) {
synchronized {
........
if (isMaster) {
trackMV = new TrackMultipleValues
trackMV.setDaemon(true)
trackMV.start()
// Set masterHostAddress to the master's IP address for the slaves to read
System.setProperty("spark.MultiTracker.MasterHostAddress", Utils.localIpAddress)
}
initialized = true
}
}
}

if (messageType == REGISTER_BROADCAST_TRACKER) {
// Receive Long
val id = ois.readObject.asInstanceOf[Long]
// Receive hostAddress and listenPort
val gInfo = ois.readObject.asInstanceOf[SourceInfo]

/ Add to the map
valueToGuideMap.synchronized {
valueToGuideMap += (id -> gInfo)
}

// Send dummy ACK
oos.writeObject(-1)
oos.flush()
} else if (messageType == UNREGISTER_BROADCAST_TRACKER) {
// Receive Long
val id = ois.readObject.asInstanceOf[Long]

// Remove from the map
valueToGuideMap.synchronized {
valueToGuideMap(id) = SourceInfo("", SourceInfo.TxOverGoToDefault)
}

// Send dummy ACK
oos.writeObject(-1)
oos.flush()
} else if (messageType == FIND_BROADCAST_TRACKER) {
// Receive Long
val id = ois.readObject.asInstanceOf[Long]

var gInfo =
if (valueToGuideMap.contains(id)) valueToGuideMap(id)
else SourceInfo("", SourceInfo.TxNotStartedRetry)

// Send reply back
oos.writeObject(gInfo)
oos.flush()
}

在初始化完成之后，与httpbroadcast类似，treebroadcast的sparkcontext的实例会通过broadcast方法，开始广播变量，期间会在master节点开启如下两个服务：
1. GuideMultipleRequests服务，保存集群中每个work节点的存有的broadcast变量信息，包括变量block个数，大小等；work节点在receivebroadcast时，会首先请求该服务，根据broadcast-id，获得相应的SourceInfo（master节点有相应的策略选择拥有该SourceInfo节点），其中包括该SourceInfo所在的hostaddress以及listenport；master节点会将该work节点的SourceInfo加入listOfSources；
2. ServeMultipleRequests服务，会在集群中的每一个节点启动，用以broadcast变量的block块的同步；work节点在获得SourceInfo之后，会向对应的节点下载block数据；由于节点既分发数据又下载数据，需要对正在同步的block加锁来确保数据的一致性；

private def sendObject() {
// Wait till receiving the SourceInfo from Master
while (totalBlocks == -1) {
totalBlocksLock.synchronized { totalBlocksLock.wait() }
}

for (i <- sendFrom until sendUntil) {
while (i == hasBlocks) {
hasBlocksLock.synchronized { hasBlocksLock.wait() }
}
}
..........
}

receiveSingleTransmission

for (i <- hasBlocks until totalBlocks) {
val recvStartTime = System.currentTimeMillis
val bcBlock = oisSource.readObject.asInstanceOf[BroadcastBlock]
val receptionTime = (System.currentTimeMillis - recvStartTime)

logDebug("Received block: " + bcBlock.blockID + " from " + sourceInfo + " in " + receptionTime + " millis.")

arrayOfBlocks(hasBlocks) = bcBlock
hasBlocks += 1

// Set to true if at least one block is received
receptionSucceeded = true
hasBlocksLock.synchronized { hasBlocksLock.notifyAll() }
}

最后简单介绍一下，master节点selectSuitableSource的策略：每一个work节点connect主节点的时候，都会被加入listOfSources，master节点按照FIFO的优先策略来选择source供work节点下载broadcast变量；首先只有master节点具有全部的数据，master节点选为源节点，每一个work节点从源节点同步数据，每连接上一个work节点，当前的源节点的leechers数加一，当源节点的leechers大于MaxDegree的时候，则停止该源节点，开始选择后来加入的work节点作为源节点，以此类推；当某个废弃的源节点完全同步成功所有的block后，该节点的leechers数减一，同时append入listOfSources，显然该节点的优先级别更高；

private def selectSuitableSource(skipSourceInfo: SourceInfo): SourceInfo = {
var maxLeechers = -1
var selectedSource: SourceInfo = null

listOfSources.foreach { source =>
if ((source.hostAddress != skipSourceInfo.hostAddress ||
source.listenPort != skipSourceInfo.listenPort) &&
source.currentLeechers < MultiTracker.MaxDegree &&
source.currentLeechers > maxLeechers) {
selectedSource = source
maxLeechers = source.currentLeechers
}
}

// Update leecher count
selectedSource.currentLeechers += 1
return selectedSource
}