先看一下原理性的文章：http://jerryshao.me/architecture/2013/10/08/spark-storage-module-analysis/ ，http://jerryshao.me/architecture/2013/10/08/spark-storage-module-analysis/  , 另外，spark的存储使用了Segment File的概念（http://en.wikipedia.org/wiki/Segmented_file_transfer ），概括的说，它是把文件划分成多个段，分别存储在不同的服务器上；在读取的时候，同时从这些服务器上读取。（这也是BT的基础）。

之前分析shuffle的调用关系的时候，其实已经包含了很多的BlockManager的流程，但还是有必要系统的看一遍它的代码。

getLocalFromDisk这个函数，是前面看shuffleManager的终点，但却是BlockManager的起点。即使是到远端获取block的操作，也是发送一个消息到远端服务器上执行getLocalFromDisk，然后再把结果发送回来。

->diskStore.getValues(blockId, serializer)

============================BlockManager============================

===========================end=================================

再次引用这个图：多个map可以对应一个文件，其中每个map对应文件中的某些段。这样做是为了减少文件数量。

/* Class for returning a fetched block and associated metrics. */private[spark] class BlockResult(val data: Iterator[Any],    readMethod: DataReadMethod.Value,    bytes: Long) {val inputMetrics = new InputMetrics(readMethod)inputMetrics.bytesRead = bytes}

private[spark] class BlockManager(    executorId: String,    actorSystem: ActorSystem,    val master: BlockManagerMaster,    defaultSerializer: Serializer,    maxMemory: Long,    val conf: SparkConf,    securityManager: SecurityManager,    mapOutputTracker: MapOutputTracker,    shuffleManager: ShuffleManager)extends BlockDataProvider with Logging {

shuffle状态，主要包含了unusedFileGroups、allFileGroups两个属性，记录当前已经使用和未使用的ShuffleFileGroup

/** * Contains all the state related to a particular shuffle. This includes a pool of unused * ShuffleFileGroups, as well as all ShuffleFileGroups that have been created for the shuffle. */private class ShuffleState(val numBuckets: Int) {val nextFileId = new AtomicInteger(0)val unusedFileGroups = new ConcurrentLinkedQueue[ShuffleFileGroup]()val allFileGroups = new ConcurrentLinkedQueue[ShuffleFileGroup]()

/**   * The mapIds of all map tasks completed on this Executor for this shuffle.   * NB: This is only populated if consolidateShuffleFiles is FALSE. We don't need it otherwise.   */  val completedMapTasks = new ConcurrentLinkedQueue[Int]()}

shuffleStates 是一个基于时间戳的hash table 

private val shuffleStates = new TimeStampedHashMap[ShuffleId, ShuffleState]

private val metadataCleaner =new MetadataCleaner(MetadataCleanerType.SHUFFLE_BLOCK_MANAGER, this.cleanup, conf)

Used by sort-based shuffle： shuffle结束时将结果注册到shuffleStates

/** * Register a completed map without getting a ShuffleWriterGroup. Used by sort-based shuffle * because it just writes a single file by itself. */def addCompletedMap(shuffleId: Int, mapId: Int, numBuckets: Int): Unit = {shuffleStates.putIfAbsent(shuffleId, new ShuffleState(numBuckets))val shuffleState = shuffleStates(shuffleId)  shuffleState.completedMapTasks.add(mapId)}

将自己注册给master 

/** * Initialize the BlockManager. Register to the BlockManagerMaster, and start the * BlockManagerWorker actor. */private def initialize(): Unit = {  master.registerBlockManager(blockManagerId, maxMemory, slaveActor)  BlockManagerWorker.startBlockManagerWorker(this)}

从本地磁盘获取一个block数据。为了方便使用

/** * A short-circuited method to get blocks directly from disk. This is used for getting * shuffle blocks. It is safe to do so without a lock on block info since disk store * never deletes (recent) items. */def getLocalFromDisk(blockId: BlockId, serializer: Serializer): Option[Iterator[Any]] = {diskStore.getValues(blockId, serializer).orElse {throw new BlockException(blockId, s"Block $blockId not found on disk, though it should be")  }}

ShuffleWriterGroup：每个shuffleMapTask都有一组shuffleWriter，它给每个reducer分配了一个writer。当前只有HashShufflle使用了，唯一一个实例化是在forMapTask返回的，给HashShuffleWriter的shuffle属性使用：

/** A group of writers for a ShuffleMapTask, one writer per reducer. */private[spark] trait ShuffleWriterGroup {val writers: Array[BlockObjectWriter]

/** @param success Indicates all writes were successful. If false, no blocks will be recorded. */  def releaseWriters(success: Boolean)}

/** * Manages assigning disk-based block writers to shuffle tasks. Each shuffle task gets one file * per reducer (this set of files is called a ShuffleFileGroup). * * As an optimization to reduce the number of physical shuffle files produced, multiple shuffle * blocks are aggregated into the same file. There is one "combined shuffle file" per reducer * per concurrently executing shuffle task. As soon as a task finishes writing to its shuffle * files, it releases them for another task. * Regarding the implementation of this feature, shuffle files are identified by a 3-tuple: *   - shuffleId: The unique id given to the entire shuffle stage. *   - bucketId: The id of the output partition (i.e., reducer id) *   - fileId: The unique id identifying a group of "combined shuffle files." Only one task at a *       time owns a particular fileId, and this id is returned to a pool when the task finishes. * Each shuffle file is then mapped to a FileSegment, which is a 3-tuple (file, offset, length) * that specifies where in a given file the actual block data is located. * * Shuffle file metadata is stored in a space-efficient manner. Rather than simply mapping * ShuffleBlockIds directly to FileSegments, each ShuffleFileGroup maintains a list of offsets for * each block stored in each file. In order to find the location of a shuffle block, we search the * files within a ShuffleFileGroups associated with the block's reducer. */// TODO: Factor this into a separate class for each ShuffleManager implementationprivate[spark]class ShuffleBlockManager(blockManager: BlockManager,                          shuffleManager: ShuffleManager) extends Logging {

ShuffleFileGroup是一组文件，每个reducer对应一个。每个map将会对应一个这个文件（但多个map可以对应一个文件）。多个map对应一个文件时，它们写入是分段写入的（mapId，ReduceId）通过getFileSegmentFor函数获取到这个块的内容

privateobject /**   * .   * .   */  private class val Int, val Int, val private var numBlocksInt 0

    /**     *  For instance,     * if mapId 5 is the first block in each file, mapIdToIndex(5) = 0.     */    private val mapIdToIndex new Int, Int/**     * Stores consecutive offsets and lengths of blocks into each reducer file, ordered by     * position in the file.     * Note:      * .     */    private val blockOffsetsByReducer fillLongnew Longprivate val blockLengthsByReducer fillLongnew Longdef applyIntdef recordMapOutputInt, Long, LongassertmapIdToIndexnumBlocks      numBlocks 1      for 0 blockOffsetsByReducerblockLengthsByReducer/** Returns the FileSegment associated with the given map task, or None if no entry exists. */    def getFileSegmentForInt, Intval val blockOffsetsByReducerval blockLengthsByReducerval mapIdToIndex, 1if 0val val Somenew , , else 

来自为知笔记(Wiz)