spark BlockManager粗讲

maxMemory = memoryManager.maxOnHeapStorageMemory + memoryManager.maxOffHeapStorageMemory

BlockManagerMasterEndpoint:存在于Driver端，主要对Executor和BlockManager的关系，BlockManager和Block的关系等进行管理。并作为一个RpcEndpoint接受各类消息事件进行处理。
BlockManager：运行在各个节点包括Driver和Executor，提供接口来支持发送和接受blocks处理。
BlockManagerStorageEndpoint：RpcEndpoint来响应Master发来的各种命令，如删除block，删除shuffle，删除Broadcast等。
BlockInfoManager：BlockManager中负责管理blocks元数据和block的锁管理
BlockManagerMaster：BlockManager中负责和BlockManagerMasterEndpoint进行通信处理DiskBlockManager：管理存放在磁盘上的Block，如block和磁盘存储路径的关系。
DiskStore：磁盘存储，负责对block存放到磁盘上进行操作处理
MemoryManager：内存块管理，这里为了方便灵活使用内存，把存储使用的内存和运行使用的内存都通过该类来实现了，这个后面会单独介绍。这里管理的内存包括堆内存和堆外内存
MemoryStore：存储block到内存中，可以为java对象和序列化后的bytebuffer
SerializerManager: 序列化处理，负责block的序列化和反序列化处理
BlockTransferService：块传输服务，如提交一个块到一个远端节点
ShuffleManager：shuffle管理，主要支持计算过程中的shuffle处理
MapOutputTracker：map任务输出跟踪管理

MemoryStore#putIteratorAsValues[T](blockId: BlockId,
    values: Iterator[T],
    classTag: ClassTag[T]): Either[PartiallyUnrolledIterator[T], Long]

unroll block：展开迭代器
store unrolled to storage memory：将展开后的数据存入 storage 内存

DiskBlockManager 
创建并维护逻辑block和block落地的物理文件的映射关系。一个逻辑block通过它的BlockId的name属性映射到具体的文件。
CountingWritableChannel
它主要对sink做了包装，在写入sink的同时，还记录向sink写的数据的总量
ManagedBuffer
ManagedBuffer以字节数组的方式提供数据，并且数据不能被修改。 它的实现应该制定数据是怎么提供的。具体的缓冲区实现可能不被JVM垃圾收回收器管理。
ManagedBuffer有三种具体的实现
   1. FileSegmentManagedBuffer： 文件的一部分支持数据。
   2. NioManagedBuffer: NIO ByteBuffer支持数据。
   3. NettyManageBuffer: Netty ByteBuf支持数据。
EncryptedManagedBuffer
它是一个适配器，它将几乎所以转换的请求委托给了 blockData,它的父类是lockData
BlockData
它是一个接口，它定义了存储方式以及如何提供不同的方式来读去底层的block 数据。
DiskBlockData
该类主要用于将磁盘中的block文件转换为指定的流或对象。
EncryptedBlockData
这个类主要是用于加密的block磁盘文件转换为特定的流或对象。
ByteBufferBlockData
这个类主要是用于内存中的block数据转换为指定的流或对象
ReferenceCounted
这是netty包下的一个接口。它是一个引用计数对象，需要显示调用deallocation。ReferenceCounted对象实例化时，引用计数设为1，调用retain方法增加引用计数，release方法则释放引用计数。
FileRegion
它也是netty下的一个包，FileRegion数据通过支持零拷贝的channel将数据传输到目标channel。
AbstractReferenceCounted
这个类是通过一个变量来记录引用的增加或减少情况。
AbstractFileRegion
AbstractFileRegion 继承了AbstractReferenceCounted， 但他还是一个抽象类，只是实现了部分的功能。
DefaultFileRegion
它通过 RandomeAccessFile 获取 可以支持随机访问 FileChannelImpl 的FileChannel，然后根据相对位置计算出绝对位置以及需要传输的字节总大小，最后将数据传输到target。
其引用计数的处理调用其父类 AbstractReferenceCounted的对应方法。
ReadableChannelFileRegion
其内部的buffer 的大小时 64KB，_traferred 变量记录了已经传输的字节数量。ReadableByteChannel 是按顺序读的，所以pos参数没有用。
DiskStore 
它就是用来保存block 到磁盘的。

1. 移除executor，有同步和异步两种方案，这两个方法只会在driver端使用。如下：removeExecutor,removeExecutorAsync
2. 向driver注册blockmanager，方法：registerBlockManager
3. 更新block信息,方法：updateBlockInfo
4. 向driver请求获取block对应的 location信息，方法：getLocations
5. 向driver 请求获得集群中所有的 blockManager的信息，方法：getPeers
6. 向driver 请求executor endpoint ref 对象,方法：getExecutorEndpointRef
7. 移除block、RDD、shuffle、broadcast,方法：removeBlock，removeRDD,removeShuffle，removeBroadcast
8. 向driver 请求获取每一个BlockManager内存状态,方法：getMemoryStatus
9. 向driver请求获取磁盘状态,方法：getStorageStatus
10. 向driver请求获取block状态，方法：getBlockStatus
11. 是否有匹配的block,方法：getMatchingBlockIds
12.检查是否缓存了block,方法：hasCachedBlocks，依赖于teil方法

1. init方法用于初始化ShuffleClient，需要指定executor 的appId
2. fetchBlocks 用于异步从另一个节点请求获取blocks，参数解释如下：
      host – the host of the remote node.
      port – the port of the remote node.
      execId – the executor id.
      blockIds – block ids to fetch.
      listener – the listener to receive block fetching status.
      downloadFileManager – DownloadFileManager to create and clean temp files. If it's not null, the remote blocks will be streamed into temp shuffle files to reduce the memory usage, otherwise, they will be kept in memory.
3. shuffleMetrics 用于记录shuffle相关的metrics信息

init 方法，它额外提供了使用BlockDataManager初始化的方法，方便从本地获取block或者将block存入本地。
close：关闭ShuffleClient
port：服务正在监听的端口
hostname：服务正在监听的hostname
fetchBlocks 跟继承类一样，没有实现，由于继承关系可以不写。
uploadBlocks：上传block到远程节点，返回一个future对象
fetchBlockSync：同步抓取远程节点的block，直到block数据获取成功才返回
uploadBlockSync 方法：同步上传信息，直到上传成功才结束。

1. 向driver端trackerEndpoint 发送消息，askTracker
2. excutor 获取每一个shuffle中task 需要读取的范围的 block信息，partition范围包头不包尾。getMapSizeByExcutorId
3. 删除指定的shuffle的状态信息,unregisterShuffle
4. 停止服务,stop

BlockManager主要功能有
    向Driver注册当前的BlockManger
    向Driver上报所管理的数据块信息
    从本地获取序列化/非序列化数据块的方法
    保存数据块到本地
    从Driver获取集群中持有某个数据块的节点信息
    从其他节点获取数据块的方法
    注册任务，获取/释放数据块上的锁
    将所持有的数据块复制到其他节点

1.shuffle客户端ShuffleClient;
  2.BlockManagerMaster(对存在于所有Executor上的BlockManager统一管理)
  3.磁盘块管理器DiskBlockManager，主要是用来映射blockId和磁盘文件的关系，维护blockID的锁，在读或写数据的时候，都是根据blockId去读取或创建一个文件;
  4.内存存储MemoryStore；
  5.磁盘存储DiskStore;
  6.非广播Block清理器metadataCleanner和广播Block清理器broadcastCleaner;
  7.压缩算法实现CompressionCodee。

def initialize(appId: String): Unit = {
    //初始化BlockTransferService，其实是它的子类NettyBlockTransferService是下了init方法，
    //该方法的作用就是初始化传输服务，通过传输服务可以从不同的节点上拉取Block数据
    blockTransferService.init(this)
    shuffleClient.init(appId)
    //设置block的复制分片策略，由spark.storage.replication.policy指定
    blockReplicationPolicy = {
      val priorityClass = conf.get(
        "spark.storage.replication.policy", classOf[RandomBlockReplicationPolicy].getName)
      val clazz = Utils.classForName(priorityClass)
      val ret = clazz.newInstance.asInstanceOf[BlockReplicationPolicy]
      logInfo(s"Using $priorityClass for block replication policy")
      ret
    }
    //根据给定参数为对对应的Executor封装一个BlockManagerId对象（块存储的唯一标识）
    //executorID：executor的Id，blockTransferService.hostName：传输Block数据的服务的主机名
    //blockTransferService.port：传输Block数据的服务的主机名
    val id = BlockManagerId(executorId, blockTransferService.hostName, blockTransferService.port, None)
    //调用BlockManagerMaster的registerBlockManager方法向Driver上的BlockManagerMaster注册
    val idFromMaster = master.registerBlockManager(
      id,
      maxMemory,
      slaveEndpoint)
    //更新BlockManagerId
    blockManagerId = if (idFromMaster != null) idFromMaster else id
    //判断是否开了外部shuffle服务
    shuffleServerId = if (externalShuffleServiceEnabled) {
      logInfo(s"external shuffle service port = $externalShuffleServicePort")
      BlockManagerId(executorId, blockTransferService.hostName, externalShuffleServicePort)
    } else {
      blockManagerId
    }
    // 如果开启了外部shuffle服务，并且该节点是Driver的话就调用registerWithExternalShuffleServer方法
    //将BlockManager注册在本地
    if (externalShuffleServiceEnabled && !blockManagerId.isDriver) {
      registerWithExternalShuffleServer()
    }
    logInfo(s"Initialized BlockManager: $blockManagerId")
  }

def getBytes(blockId: BlockId): Option[ChunkedByteBuffer] = {
    //通过blockId从LinkedHashMap内存中获取entry对象
    val entry = entries.synchronized { entries.get(blockId) }
    entry match {
      case null => None
      case e: DeserializedMemoryEntry[_] =>
        throw new IllegalArgumentException("should only call getBytes on serialized blocks")
      //通过模式匹配，验证entry类型，并提取序列化类中的数据信息
      case SerializedMemoryEntry(bytes, _, _) => Some(bytes)
    }
  }

override def fetchBlocks(
    //需要传入ip和port
      host: String,
      port: Int,
      execId: String,
      blockIds: Array[String],
      listener: BlockFetchingListener,
      tempFileManager: DownloadFileManager): Unit = {
    if (logger.isTraceEnabled) {
      logger.trace(s"Fetch blocks from $host:$port (executor id $execId)")
    }
    try {
      val maxRetries = transportConf.maxIORetries()
      //创建远程块数据下载的启动模块，并实现启动方法
      val blockFetchStarter = new RetryingBlockFetcher.BlockFetchStarter {
        override def createAndStart(blockIds: Array[String],
            listener: BlockFetchingListener): Unit = {
          try {
            //创建传输客户端，用于连接远程节点
            val client = clientFactory.createClient(host, port, maxRetries > 0)
            //启动一对一的数据块获取
            new OneForOneBlockFetcher(client, appId, execId, blockIds, listener,
              transportConf, tempFileManager).start()
          } catch {
            case e: IOException =>
              Try {
                driverEndPointRef.askSync[Boolean](IsExecutorAlive(execId))
              } match {
                case Success(v) if v == false =>
                  throw new ExecutorDeadException(s"The relative remote executor(Id: $execId)," +
                    " which maintains the block data to fetch is dead.")
                case _ => throw e
              }
          }
        }
      }
       //只有最大重试次数大于0，才会走封装的重试类，否则直接是开启一对一下载
      if (maxRetries > 0) {
        // Note this Fetcher will correctly handle maxRetries == 0; we avoid it just in case there's
        // a bug in this code. We should remove the if statement once we're sure of the stability.
        new RetryingBlockFetcher(transportConf, blockFetchStarter, blockIds, listener).start()
      } else {
        blockFetchStarter.createAndStart(blockIds, listener)
      }
    } catch {
      case e: Exception =>
        logger.error("Exception while beginning fetchBlocks", e)
        blockIds.foreach(listener.onBlockFetchFailure(_, e))
    }
  }

public void start() {
    //想远程节点发送rpc请求，并在回调函数中监听远程节点的响应
    client.sendRpc(message.toByteBuffer(), new RpcResponseCallback() {
      @Override
      public void onSuccess(ByteBuffer response) {
        try {
          //创建流处理器处理远程节点返回的数据
          streamHandle = (StreamHandle) BlockTransferMessage.Decoder.fromByteBuffer(response);
          logger.trace("Successfully opened blocks {}, preparing to fetch chunks.", streamHandle);
          // 遍历获取远程节点提供的block数据
          for (int i = 0; i < streamHandle.numChunks; i++) {
            if (downloadFileManager != null) {
              client.stream(OneForOneStreamManager.genStreamChunkId(streamHandle.streamId, i),
                new DownloadCallback(i));
            } else {
              client.fetchChunk(streamHandle.streamId, i, chunkCallback);
            }
          }
        } catch (Exception e) {
          logger.error("Failed while starting block fetches after success", e);
          failRemainingBlocks(blockIds, e);
        }
      }
      @Override
      public void onFailure(Throwable e) {
        logger.error("Failed while starting block fetches", e);
        failRemainingBlocks(blockIds, e);
      }
    });
  }

override def uploadBlock(
      hostname: String,
      port: Int,
      execId: String,
      blockId: BlockId,
      blockData: ManagedBuffer,
      level: StorageLevel,
      classTag: ClassTag[_]): Future[Unit] = {
    val result = Promise[Unit]()
    val client = clientFactory.createClient(hostname, port)
    // 序列化元数据
    val metadata = JavaUtils.bufferToArray(serializer.newInstance().serialize((level, classTag)))
    // 如果上传的数据量超过一定量则通过流式处理器上传
    val asStream = blockData.size() > conf.get(config.MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM)
    //上传成功或者失败的回调函数
    val callback = new RpcResponseCallback {
      override def onSuccess(response: ByteBuffer): Unit = {
        logTrace(s"Successfully uploaded block $blockId${if (asStream) " as stream" else ""}")
        result.success((): Unit)
      }
      override def onFailure(e: Throwable): Unit = {
        logError(s"Error while uploading $blockId${if (asStream) " as stream" else ""}", e)
        result.failure(e)
      }
    }
    //根据是否需要流处理进而走不通的逻辑
    if (asStream) {
      //如果是流式处理，则封装流处理器，然后分批上传
      val streamHeader = new UploadBlockStream(blockId.name, metadata).toByteBuffer
      client.uploadStream(new NioManagedBuffer(streamHeader), blockData, callback)
    } else {
      // 如果数据量比较小，则一次性传输完，而不需要分批处理
      val array = JavaUtils.bufferToArray(blockData.nioByteBuffer())
      client.sendRpc(new UploadBlock(appId, execId, blockId.name, metadata, array).toByteBuffer,
        callback)
    }
    result.future
  }