Spark Block存储管理分析

   <p>Apache Spark中，对Block的查询、存储管理，是通过唯一的Block ID来进行区分的。所以，了解Block ID的生成规则，能够帮助我们了解Block查询、存储过程中是如何定位Block以及如何处理互斥存储/读取同一个Block的。可以想到，同一个Spark Application，以及多个运行的Application之间，对应的Block都具有唯一的ID，通过代码可以看到，BlockID包括：RDDBlockId、ShuffleBlockId、ShuffleDataBlockId、ShuffleIndexBlockId、BroadcastBlockId、TaskResultBlockId、TempLocalBlockId、TempShuffleBlockId这8种ID，可以详见如下代码定义：</p>    <pre>  <code class="language-java">@DeveloperApi  case class RDDBlockId(rddId: Int, splitIndex: Int) extends BlockId {    override def name: String = "rdd_" + rddId + "_" + splitIndex  }    // Format of the shuffle block ids (including data and index) should be kept in sync with  // org.apache.spark.network.shuffle.ExternalShuffleBlockResolver#getBlockData().  @DeveloperApi  case class ShuffleBlockId(shuffleId: Int, mapId: Int, reduceId: Int) extends BlockId {    override def name: String = "shuffle_" + shuffleId + "_" + mapId + "_" + reduceId  }    @DeveloperApi  case class ShuffleDataBlockId(shuffleId: Int, mapId: Int, reduceId: Int) extends BlockId {    override def name: String = "shuffle_" + shuffleId + "_" + mapId + "_" + reduceId + ".data"  }    @DeveloperApi  case class ShuffleIndexBlockId(shuffleId: Int, mapId: Int, reduceId: Int) extends BlockId {    override def name: String = "shuffle_" + shuffleId + "_" + mapId + "_" + reduceId + ".index"  }    @DeveloperApi  case class BroadcastBlockId(broadcastId: Long, field: String = "") extends BlockId {    override def name: String = "broadcast_" + broadcastId + (if (field == "") "" else "_" + field)  }    @DeveloperApi  case class TaskResultBlockId(taskId: Long) extends BlockId {    override def name: String = "taskresult_" + taskId  }    @DeveloperApi  case class StreamBlockId(streamId: Int, uniqueId: Long) extends BlockId {    override def name: String = "input-" + streamId + "-" + uniqueId  }    /** Id associated with temporary local data managed as blocks. Not serializable. */  private[spark] case class TempLocalBlockId(id: UUID) extends BlockId {    override def name: String = "temp_local_" + id  }    /** Id associated with temporary shuffle data managed as blocks. Not serializable. */  private[spark] case class TempShuffleBlockId(id: UUID) extends BlockId {    override def name: String = "temp_shuffle_" + id  }</code></pre>    <p>我们以RDDBlockId的生成规则为例，它是以前缀字符串“rdd_”为前缀、分配的全局RDD ID、下划线“_”、Partition ID这4部分拼接而成，因为RDD ID是唯一的，所以最终构造好的RDDBlockId对应的字符串就是唯一的。如果该Block存在，查询可以唯一定位到该Block，存储也不会出现覆盖其他RDDBlockId的问题。</p>    <p>下面，我们通过分析MemoryStore、DiskStore、BlockManager、BlockInfoManager这4个最核心的与Block管理相关的实现类，来理解Spark对Block的管理。全文中，我们主要针对RDDBlockId对应的Block数据的处理、存储、查询、读取，来分析Block的管理。</p>    <p>MemoryStore</p>    <p>先说明一下MemoryStore，它主要用来在内存中存储Block数据，可以避免重复计算同一个RDD的Partition数据。一个Block对应着一个RDD的一个Partition的数据。当StorageLevel设置为如下值时，都会可能会需要使用MemoryStore来存储数据：</p>    <pre>  <code class="language-java">MEMORY_ONLY  MEMORY_ONLY_2  MEMORY_ONLY_SER  MEMORY_ONLY_SER_2  MEMORY_AND_DISK  MEMORY_AND_DISK_2  MEMORY_AND_DISK_SER  MEMORY_AND_DISK_SER_2  OFF_HEAP</code></pre>    <p>所以，MemoryStore提供对Block数据的存储、读取等操作API，MemoryStore也提供了多种存储方式，下面详细说明每种方式。</p>    <ul>     <li>以序列化格式保存Block数据</li>    </ul>    <pre>  <code class="language-java">def putBytes[T: ClassTag](        blockId: BlockId,        size: Long,        memoryMode: MemoryMode,        _bytes: () => ChunkedByteBuffer): Boolean</code></pre>    <p>首先，通过MemoryManager来申请Storage内存，调用putBytes方法，会根据size大小去申请Storage内存，如果申请成功，则会将blockId对应的Block数据保存在内部的LinkedHashMap[BlockId, MemoryEntry[_]]映射表中，然后以SerializedMemoryEntry这种序列化的格式存储，实际SerializedMemoryEntry就是简单指向Buffer中数据的引用对象：</p>    <pre>  <code class="language-java">private case class SerializedMemoryEntry[T](      buffer: ChunkedByteBuffer,      memoryMode: MemoryMode,      classTag: ClassTag[T]) extends MemoryEntry[T] {    def size: Long = buffer.size  }</code></pre>    <p>如果无法申请到size大小的Storage内存，则存储失败，对于出现这种失败的情况，需要使用MemoryStore存储API的调用者去处理异常情况。</p>    <ul>     <li>基于记录迭代器，以反序列化Java对象形式保存Block数据</li>    </ul>    <pre>  <code class="language-java">private[storage] def putIteratorAsValues[T](        blockId: BlockId,        values: Iterator[T],        classTag: ClassTag[T]): Either[PartiallyUnrolledIterator[T], Long]</code></pre>    <p>这种方式，调用者希望将Block数据记录以反序列化的方式保存在内存中，如果内存中能放得下，则返回最终Block数据记录的大小，否则返回一个PartiallyUnrolledIterator[T]迭代器，其中对应如下2种情况：</p>    <p>第一种，Block数据记录能够完全放到内存中：和前面的方式类似，能够全部放到内存，但是不同的是，这种方式对应的数据格式是反序列化的Java对象格式，对应实现类DeserializedMemoryEntry[T]，它也会被直接存放到MemoryStore内部的LinkedHashMap[BlockId, MemoryEntry[_]]映射表中。DeserializedMemoryEntry[T]类定义如下所示：</p>    <pre>  <code class="language-java">private case class DeserializedMemoryEntry[T](      value: Array[T],      size: Long,      classTag: ClassTag[T]) extends MemoryEntry[T] {    val memoryMode: MemoryMode = MemoryMode.ON_HEAP  }</code></pre>    <p>它与SerializedMemoryEntry都是MemoryEntry[T]的子类，所有被放到同一个映射表LinkedHashMap[BlockId, MemoryEntry[_]] entries中。</p>    <p>另外，也存在这种可能，通过MemoryManager申请的Unroll内存大小大于该Block打开需要的内存，则会返回如下结果对象：</p>    <pre>  <code class="language-java">Left(new PartiallyUnrolledIterator( this,  unrollMemoryUsedByThisBlock,             unrolled = arrayValues.toIterator,  rest = Iterator.empty))</code></pre>    <p>上面unrolled = arrayValues.toIterator，rest = Iterator.empty，表示在内存中可以打开迭代器中全部的数据记录，打开对象类型为DeserializedMemoryEntry[T]。</p>    <p>第二种，Block数据记录只能部分放到内存中：也就是说Driver或Executor上的内存有限，只可以放得下部分记录，另一部分记录内存中放不下。values记录迭代器对应的全部记录数据无法完全放在内存中，所以为了保证不发生OOM异常，首选会调用MemoryManager的acquireUnrollMemory方法去申请Unroll内存，如果可以申请到，在迭代values的过程中，需要累加计算打开（Unroll）的记录对象大小之和，使其大小不能大于申请到的Unroll内存，直到还有一部分记录无法放到申请的Unroll内存中。 最后，返回的结果对象如下所示：</p>    <pre>  <code class="language-java">Left(new PartiallyUnrolledIterator(          this, unrollMemoryUsedByThisBlock, unrolled = vector.iterator, rest = values))</code></pre>    <p>上面的PartiallyUnrolledIterator中rest对应的values就是putIteratorAsValues方法传进来的迭代器参数值，该迭代器已经迭代出部分记录，放到了内存中，调用者可以继续迭代该迭代器去处理未打开（Unroll）的记录，而unrolled对应一个打开记录的迭代器。这里，PartiallyUnrolledIterator迭代器包装了vector.iterator和一个迭代出部分记录的values迭代器，调用者对PartiallyUnrolledIterator进行统一迭代能够获取到全部记录，里面包含两种类型的记录：DeserializedMemoryEntry[T]和T。</p>    <ul>     <li>基于记录迭代器，以序列化二进制格式保存Block数据</li>    </ul>    <pre>  <code class="language-java">private[storage] def putIteratorAsBytes[T](        blockId: BlockId,        values: Iterator[T],        classTag: ClassTag[T],        memoryMode: MemoryMode): Either[PartiallySerializedBlock[T], Long]</code></pre>    <p>这种方式，调用这种希望将Block数据记录以二进制的格式保存在内存中。如果内存中能放得下，则返回最终的大小，否则返回一个PartiallySerializedBlock[T]迭代器。</p>    <p>如果Block数据记录能够完全放到内存中，则以SerializedMemoryEntry[T]格式放到内存的映射表中。如果Block数据记录只能部分放到内存中，则返回如下对象：</p>    <pre>  <code class="language-java">Left(          new PartiallySerializedBlock(            this,            serializerManager,            blockId,            serializationStream,            redirectableStream,            unrollMemoryUsedByThisBlock,            memoryMode,            bbos.toChunkedByteBuffer,            values,            classTag))</code></pre>    <p>类似地，返回结果对象对调用者保持统一的迭代API视图。</p>    <p>DiskStore</p>    <p>DiskStore提供了将Block数据写入到磁盘的基本操作，它是通过DiskBlockManager来管理逻辑上Block到物理磁盘上Block文件路径的映射关系。当StorageLevel设置为如下值时，都可能会需要使用DiskStore来存储数据：</p>    <pre>  <code class="language-java">DISK_ONLY  DISK_ONLY_2  MEMORY_AND_DISK  MEMORY_AND_DISK_2  MEMORY_AND_DISK_SER  MEMORY_AND_DISK_SER_2  OFF_HEAP</code></pre>    <p>DiskBlockManager管理了每个Block数据存储位置的信息，包括从Block ID到磁盘上文件的映射关系。DiskBlockManager主要有如下几个功能：</p>    <ul>     <li>负责创建一个本地节点上的指定磁盘目录，用来存储Block数据到指定文件中</li>     <li>如果Block数据想要落盘，需要通过调用getFile方法来分配一个唯一的文件路径</li>     <li>如果想要查询一个Block是否在磁盘上，通过调用containsBlock方法来查询</li>     <li>查询当前节点上管理的全部Block文件</li>     <li>通过调用createTempLocalBlock方法，生成一个唯一Block ID，并创建一个唯一的临时文件，用来存储中间结果数据</li>     <li>通过调用createTempShuffleBlock方法，生成一个唯一Block ID，并创建一个唯一的临时文件，用来存储Shuffle过程的中间结果数据</li>    </ul>    <p>DiskStore提供的基本操作接口，与MemoryStore类似，比较简单，如下所示：</p>    <ul>     <li>通过文件流写Block数据</li>    </ul>    <p>该种方式对应的接口方法，如下所示：</p>    <pre>  <code class="language-java">def put(blockId: BlockId)(writeFunc: FileOutputStream => Unit): Unit</code></pre>    <p>参数指定Block ID，还有一个写Block数据到打开的文件流的函数，在调用put方法时，首先会从DiskBlockManager分配一个Block ID对应的磁盘文件路径，然后将数据写入到该文件中。</p>    <ul>     <li>将二进制Block数据写入文件</li>    </ul>    <p>putBytes方法实现了，将一个Buffer中的Block数据写入指定的Block ID对应的文件中，方法定义如下所示：</p>    <pre>  <code class="language-java">def putBytes(blockId: BlockId, bytes: ChunkedByteBuffer): Unit</code></pre>    <p>实际上，它是调用上面的put方法，将bytes中的Block二进制数据写入到Block文件中。</p>    <ul>     <li>从磁盘文件读取Block数据</li>    </ul>    <p>对应方法如下所示：</p>    <pre>  <code class="language-java">def getBytes(blockId: BlockId): ChunkedByteBuffer</code></pre>    <p>通过给定的blockId，获取磁盘上对应的Block文件的数据，以ChunkedByteBuffer的形式返回。</p>    <ul>     <li>删除Block文件</li>    </ul>    <p>对应的删除方法定义，如下所示：</p>    <pre>  <code class="language-java">def remove(blockId: BlockId): Boolean</code></pre>    <p>通过DiskBlockManager查找到blockId对应的Block文件，然后删除掉。</p>    <p>BlockManager</p>    <p>谈到Spark中的Block数据存储，我们很容易能够想到BlockManager，他负责管理在每个Dirver和Executor上的Block数据，可能是本地或者远程的。具体操作包括查询Block、将Block保存在指定的存储中，如内存、磁盘、堆外（Off-heap）。而BlockManager依赖的后端，对Block数据进行内存、磁盘存储访问，都是基于前面讲到的MemoryStore、DiskStore。</p>    <p>在Spark集群中，当提交一个Application执行时，该Application对应的Driver以及所有的Executor上，都存在一个BlockManager、BlockManagerMaster，而BlockManagerMaster是负责管理各个BlockManager之间通信，这个BlockManager管理集群，如下图所示：</p>    <p style="text-align:center"><img src="https://simg.open-open.com/show/152796fc65effa354430bb7bf510fa8d.png"></p>    <p>关于一个Application运行过程中Block的管理，主要是基于该Application所关联的一个Driver和多个Executor构建了一个Block管理集群：Driver上的(BlockManagerMaster, BlockManagerMasterEndpoint)是集群的Master角色，所有Executor上的(BlockManagerMaster, RpcEndpointRef)作为集群的Slave角色。当Executor上的Task运行时，会查询对应的RDD的某个Partition对应的Block数据是否处理过，这个过程中会触发多个BlockManager之间的通信交互。我们以ShuffleMapTask的运行为例，对应代码如下所示：</p>    <pre>  <code class="language-java">override def runTask(context: TaskContext): MapStatus = {      // Deserialize the RDD using the broadcast variable.      val deserializeStartTime = System.currentTimeMillis()      val ser = SparkEnv.get.closureSerializer.newInstance()      val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])](        ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)      _executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime        var writer: ShuffleWriter[Any, Any] = null      try {        val manager = SparkEnv.get.shuffleManager        writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context)        writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]]) // 处理RDD的Partition的数据        writer.stop(success = true).get      } catch {        case e: Exception =>          try {            if (writer != null) {              writer.stop(success = false)            }          } catch {            case e: Exception =>              log.debug("Could not stop writer", e)          }          throw e      }    }</code></pre>    <p>一个RDD的Partition对应一个ShuffleMapTask，一个ShuffleMapTask会在一个Executor上运行，它负责处理RDD的一个Partition对应的数据，基本处理流程，如下所示：</p>    <ol>     <li>根据该Partition的数据，创建一个RDDBlockId（由RDD ID和Partition Index组成），即得到一个稳定的blockId（如果该Partition数据被处理过，则可能本地或者远程Executor存储了对应的Block数据）。</li>     <li>先从BlockManager获取该blockId对应的数据是否存在，如果本地存在（已经处理过），则直接返回Block结果（BlockResult）；否则，查询远程的Executor是否已经处理过该Block，处理过则直接通过网络传输到当前Executor本地，并根据StorageLevel设置，保存Block数据到本地MemoryStore或DiskStore，同时通过BlockManagerMaster上报Block数据状态（通知Driver当前的Block状态，亦即，该Block数据存储在哪个BlockManager中）。</li>     <li>如果本地及远程Executor都没有处理过该Partition对应的Block数据，则调用RDD的compute方法进行计算处理，并将处理的Block数据，根据StorageLevel设置，存储到本地MemoryStore或DiskStore。</li>     <li>根据ShuffleManager对应的ShuffleWriter，将返回的该Partition的Block数据进行Shuffle写入操作</li>    </ol>    <p>下面，我们基于上面逻辑，详细分析在这个处理过程中重要的交互逻辑：</p>    <ul>     <li>根据RDD获取一个Partition对应数据的记录迭代器</li>    </ul>    <p>用户提交的Spark Application程序，会设置对应的StorageLevel，所以设置与不设置对该处理逻辑有一定影响，具有两种情况，如下图所示：</p>    <p style="text-align:center"><img src="https://simg.open-open.com/show/3933c8961cbdfd48299394a4cebe985a.png"></p>    <p>如果用户程序设置了StorageLevel，可能该Partition的数据已经处理过，那么对应的处理结果Block数据可能已经存储。一般设置的StorageLevel，或者将Block存储在内存中，或者存储在磁盘上，这里会尝试调用getOrElseUpdate()方法获取对应的Block数据，如果存在则直接返回Block对应的记录的迭代器实例，就不需要重新计算了，如果没有找到对应的已经处理过的Block数据，则调用RDD的compute()方法进行处理，处理结果根据StorageLevel设置，将Block数据存储在内存或磁盘上，缓存供后续Task重复使用。</p>    <p>如果用户程序没有设置StorageLevel，那么RDD对应的该Partition的数据一定没有进行处理过，即使处理过，如果没有进行Checkpointing，也需要重新计算（如果进行了Checkpointing，可以直接从缓存中获取），直接调用RDD的compute()方法进行处理。</p>    <ul>     <li>从BlockManager查询获取Block数据</li>    </ul>    <p>每个Executor上都有一个BlockManager实例，负责管理用户提交的该Application计算过程中产生的Block。很有可能当前Executor上存储在RDD对应Partition的经过处理后得到的Block数据，也有可能当前Executor上没有，但是其他Executor上已经处理过并缓存了Block数据，所以对应着本地获取、远程获取两种可能，本地获取交互逻辑如下图所示：</p>    <p style="text-align:center"><img src="https://simg.open-open.com/show/de6b844e8e88a95a44885726aeeba603.png"></p>    <p>从本地获取，根据StorageLevel设置，如果是存储在内存中，则从本地的MemoryStore中查询，存在则读取并返回；如果是存储在磁盘上，则从本地的DiskStore中查询，存在则读取并返回。本地不存在，则会从远程的Executor读取，对应的组件交互逻辑，如下图所示：</p>    <p><img src="https://simg.open-open.com/show/bffbdda27ff17efc451eb284571ca07c.png"></p>    <p>远程获取交互逻辑相对比较复杂：当前Executor上的BlockManager通过BlockManagerMaster，向远程的Driver上的BlockManagerMasterEndpoint查询对应Block ID，有哪些Executor已经保存了该Block数据，Dirver返回一个包含了该Block数据的Location列表，如果对应的Location信息与当前ShuffleMapTask执行所在Executor在同一台节点上，则会优先使用该Location，因为同一节点上的多个Executor之间传输Block数据效率更高。</p>    <p>这里需要说明的是，如果对应的Block数据的StorageLevel设置为写磁盘，通过前面我们知道，DiskStore是通过DiskBlockManager进行管理存储到磁盘上的Block数据文件的，在同一个节点上的多个Executor共享相同的磁盘文件路径，相同的Block数据文件也就会被同一个节点上的多个Executor所共享。而对应MemoryStore，因为每个Executor对应独立的JVM实例，从而具有独立的Storage/Execution内存管理，所以使用MemoryStore不能共享同一个Block数据，但是同一个节点上的多个Executor之间的MemoryStore之间拷贝数据，比跨网络传输要高效的多。</p>    <p>BlockInfoManager</p>    <p>用户提交一个Spark Application程序，如果程序对应的DAG图相对复杂，其中很多Task计算的结果Block数据都有可能被重复使用，这种情况下如何去控制某个Executor上的Task线程去读写Block数据呢？其实，BlockInfoManager就是用来控制Block数据读写操作，并且跟踪Task读写了哪些Block数据的映射关系，这样如果两个Task都想去处理同一个RDD的同一个Partition数据，如果没有锁来控制，很可能两个Task都会计算并写同一个Block数据，从而造成混乱。我们分析每种情况下，BlockInfoManager是如何管理Block数据（同一个RDD的同一个Partition）读写的：</p>    <ul>     <li>第一个Task请求写Block数据</li>    </ul>    <p>这种情况下，没有其他Task写Block数据，第一个Task直接获取到写锁，并启动写Block数据到本地MemoryStore或DiskStore。如果其他写Block数据的Task也请求写锁，则该Task会阻塞，等待第一个获取写锁的Task完成写Block数据，直到第一个Task写完成，并通知其他阻塞的Task，然后其他Task需要再次获取到读锁来读取该Block数据。</p>    <ul>     <li>第一个Task正在写Block数据，其他Task请求读Block数据</li>    </ul>    <p>这种情况，Block数据没有完成写操作，其他读Block数据的Task只能阻塞，等待写Block的Task完成并通知读Task去读取Block数据。</p>    <ul>     <li>Task请求读取Block数据</li>    </ul>    <p>如果该Block数据不存在，则直接返回空，表示当前RDD的该Partition并没有被处理过。如果当前Block数据存在，并且没有其他Task在写，表示已经完成了些Block数据操作，则该Task直接读取该Block数据。</p>    <p> </p>    <p> </p>    <p> </p>    <p>来自：http://shiyanjun.cn/archives/1641.html</p>    <p> </p>