spark block读写流程分析

Posted 2020-10-15 超级核弹头

之前分析了spark任务提交以及计算的流程，本文将分析在计算过程中数据的读写过程。我们知道：spark抽象出了RDD，在物理上RDD通常由多个Partition组成，一个partition对应一个block。在driver和每个executor端，都有一个Blockmanager。Blockmanager是spark在计算过程中对block进行读写的入口，它屏蔽了在读取数据时涉及到的内存分配，从其他executor端远程获取等具体细节。接下来，本文将以读写block为主线，分析spark在计算过程中读写实际数据的流程。

 

1，计算数据写流程

1.1，从计算上来说, RDD中的一个Partition对应一个Task。在Task在taskRunner的run方法中调用task.run方法，然后根据计算结果的大小，以不同形式（直接发送或者通过blockManager）将数据发送给driver。

val value = try {
  val res = task.run(
    taskAttemptId = taskId,
    attemptNumber = attemptNumber,
    metricsSystem = env.metricsSystem)
  threwException = false
 res
}

...

// directSend = sending directly back to the driver
val serializedResult: ByteBuffer = {
  if (maxResultSize > 0 && resultSize > maxResultSize) {
    logWarning(s"Finished $taskName (TID $taskId). Result is larger than maxResultSize " +
      s"(${Utils.bytesToString(resultSize)} > ${Utils.bytesToString(maxResultSize)}), " +
      s"dropping it.")
    ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
  } else if (resultSize > maxDirectResultSize) {
    val blockId = TaskResultBlockId(taskId)
    env.blockManager.putBytes(
      blockId,
      new ChunkedByteBuffer(serializedDirectResult.duplicate()),
      StorageLevel.MEMORY_AND_DISK_SER)
    logInfo(
      s"Finished $taskName (TID $taskId). $resultSize bytes result sent via BlockManager)")
    ser.serialize(new IndirectTaskResult[Any](blockId, resultSize))
  } else {
    logInfo(s"Finished $taskName (TID $taskId). $resultSize bytes result sent to driver")
    serializedDirectResult
  }
}

1.2，在使用block形式的时候，可以看到调用了blockManager的putBytes方法，在核心实现doPutBytes中，根据存储级别和是否序列化使用memstore和diskstore中不同的不方法进行存储。顾名思义，memstore和diskstore主要就是根据存储级别将对应的block储存到内存或者磁盘。

1.3，memstore的putBytes实现如下。可以看到首先需要通过memoryManager申请存储保存当前block的内存，申请到内存后改block的数据，会以BlockId和SerializeMemoryEntry的键值对，保存在memstore的entries的对象中。关于memoryManager，当前有StaticMemoryManager和UnifiedMemoryManager两种实现。StaticMemoryManager是之前老的实现，将spark计算过程使用的存储内存和计算内存按照总大小的固定比例进行分配。UnifiedMemoryManager是2.x的默认实现，相对StaticMemoryManager，UnifiedMemoryManager中存储和计算的内存是可以动态调整的。也就是说，当计算内存紧张，储存内存空闲的时候，计算内存可以借用存储内存。反之类似。

1.4，在1.3完成当前executor机器完成当前block存储以后，当需要告诉driver时（tellMaster参数），会将该block的状态汇报给driver（reportBlockStatus），通过向dirver发送UpdateBlockInfo消息。driver接收到UpdateBlockInfo消息后，将汇报过来的相关信息保存在BlockManagerMasterEndpoint的blockManagerInfo和blockLocations中。

至此，计算过程写数据的流程完成。

2，计算数据读流程

2.1，话说在TaskRunner运行结束以后，会调用execBackend.statusUpdate，会将该任务的结束的状态通过StatusUpdate的信息发送给driver。

2.2，driver端接收到StatusUpdate消息后，最终将调用TaskResultGetter的enqueueSuccessfulTask方法。在该方法中，对于使用block（即IndirectTaskResult），最终将调用blockManager的getRemoteBytes获取该blockId对应的数据。

2.3，在blockManager的getRemoteBytes方法中，主要逻辑是获取该blockId对应的存储该blockId数据的所有机器位置，通过调用blockTransferService的fetchBlockSync获取具体数据，一旦从一个指定的位置获取到数据，则立即返回。

2.4，fetchBlockSync接着会调用具体实现NettyBlockTransferService中的fetchBlocks方法，在该方法中，将通过OneForOneBlockFetcher发送OpenBlocks消息给指定目标的blockManager，从而对应的streamId等信息，然后通过

client.fetchChunk一次获取每块的数据。

2.5，在提供数据的blockManager端（即server端），接受到消息OpenBlocks消息后，首先根据blockId通过blockManager的getBlockData方法获取对应的数据，然后将该数据和一个streamId奖励对应关系（通过streamManager调用进行）

2.6，在2.4中获取到对应的streamId后，将通过ChunkFetchRequest分块获取数据。server端接受到该消息以后，streamManager将根据streamId和chunkIndex获取对应数据，然后返回给客户端。

至此，计算过程获取block数据的流程结束。