大数据：Spark Storage(一) 集群下的区块管理.doc

大数据：Spark Storage（一） 集群下的区块管理 Storage模块 在Spark中提及最多的是RDD，而RDD所交互的数据是通过Storage来实现和管理 Storage模块整体架构 1. 存储层 在Spark里，单节点的Storage的管理是通过block来管理的，每个Block的存储可以在内存里或者在磁盘中，在BlockManager里既可以管理内存的存储，同时也管理硬盘的存储，存储的标识是通过块的ID来区分的。 2. 集群下的架构 2.1 架构 在集群下Spark的Block的管理架构使用Master-Slave模式 Master : 拥有所有block的具体信息（本地和Slave节点） Slave ： 通过master获取block的信息，并且汇报自己的信息 这里的Master并不是Spark集群中分配任务的Master，而是提交task的客户端Driver，这里并没有主备设计，因为Driver client是单点的，通常Driver client crash了，计算也没有结果了，在Storage 的集群管理中Master是由driver承担。 在Executor在运行task的时候，通过blockManager获取本地的block块，如果本地找不到，尝试通过master去获取远端的块 [java] view plain copy for (pid - Random.shuffle(Seq.range(0, numBlocks))) { val pieceId = BroadcastBlockId(id, piece + pid) logDebug(sReading 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(scorrupt remote block $pieceId of $broadcastId: + s $sum != ${checksums(pid)}) } } // We found the block from remote executors/drivers BlockManager, so put the block // in this executors BlockManager. if (!bm.putBytes(pieceId, b, StorageLevel.MEMORY_AND_DISK_SER, tellMaster = true)) { throw new SparkException( sFailed to store $pieceId of $broadcastId in local BlockManager) } blocks(pid) = b case None =