您可以捐助,支持我们的公益事业。

1元 10元 50元





认证码:  验证码,看不清楚?请点击刷新验证码 必填



  求知 文章 文库 Lib 视频 iPerson 课程 认证 咨询 工具 讲座 Model Center   Code  
会员   
   
 
     
   
 订阅
  捐助

Spark Block存储管理分析
 
来源:网络 发布于: 2017-5-11
   次浏览      
 

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,可以详见如下代码定义:

@DeveloperApi

case class RDDBlockId(rddId: Int, splitIndex: Int)

extends BlockId {
override def name: String = "rdd_" + rddId + "_" +

splitInd}

// Format of the shuffle block ids (including data

and index) should be kept in sync with
// org.apache.spark.network.shuffle.ExternalShuf

fleBlockResolver#getBlockData().
@DeveloperApi
case class ShuffleBlockId(shuffleId: Int, mapId:

Int, reduceId:Int) extends BlockId {
override def name: String = "shuffle_" + shuffleI+

"_" + mapId@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
}

我们以RDDBlockId的生成规则为例,它是以前缀字符串“rdd_”为前缀、分配的全局RDD ID、下划线“_”、Partition ID这4部分拼接而成,因为RDD ID是唯一的,所以最终构造好的RDDBlockId对应的字符串就是唯一的。如果该Block存在,查询可以唯一定位到该Block,存储也不会出现覆盖其他RDDBlockId的问题。

下面,我们通过分析MemoryStore、DiskStore、BlockManager、BlockInfoManager这4个最核心的与Block管理相关的实现类,来理解Spark对Block的管理。全文中,我们主要针对RDDBlockId对应的Block数据的处理、存储、查询、读取,来分析Block的管理。

MemoryStore

先说明一下MemoryStore,它主要用来在内存中存储Block数据,可以避免重复计算同一个RDD的Partition数据。一个Block对应着一个RDD的一个Partition的数据。当StorageLevel设置为如下值时,都会可能会需要使用MemoryStore来存储数据:

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

所以,MemoryStore提供对Block数据的存储、读取等操作API,MemoryStore也提供了多种存储方式,下面详细说明每种方式。

以序列化格式保存Block数据

def putBytes[T: ClassTag](
blockId: BlockId,
size: Long,
memoryMode: MemoryMode,
_bytes: () => ChunkedByteBuffer): Boolean

首先,通过MemoryManager来申请Storage内存,调用putBytes方法,会根据size大小去申请Storage内存,如果申请成功,则会将blockId对应的Block数据保存在内部的LinkedHashMap[BlockId, MemoryEntry[_]]映射表中,然后以SerializedMemoryEntry这种序列化的格式存储,实际SerializedMemoryEntry就是简单指向Buffer中数据的引用对象:

private case class SerializedMemoryEntry[T](
buffer: ChunkedByteBuffer,
memoryMode: MemoryMode,
classTag: ClassTag[T]) extends MemoryEntry[T] {
def size: Long = buffer.size
}

如果无法申请到size大小的Storage内存,则存储失败,对于出现这种失败的情况,需要使用MemoryStore存储API的调用者去处理异常情况。

基于记录迭代器,以反序列化Java对象形式保存Block数据

private[storage] def putIteratorAsValues[T](

blockId: BlockId,
values: Iterator[T],
classTag: ClassTag[T]): Either[PartiallyUnro

\lledIterator[T], Long]

这种方式,调用者希望将Block数据记录以反序列化的方式保存在内存中,如果内存中能放得下,则返回最终Block数据记录的大小,否则返回一个PartiallyUnrolledIterator[T]迭代器,其中对应如下2种情况:

第一种,Block数据记录能够完全放到内存中:和前面的方式类似,能够全部放到内存,但是不同的是,这种方式对应的数据格式是反序列化的Java对象格式,对应实现类DeserializedMemoryEntry[T],它也会被直接存放到MemoryStore内部的LinkedHashMap[BlockId, MemoryEntry[_]]映射表中。DeserializedMemoryEntry[T]类定义如下所示:

private case class DeserializedMemoryEntry[T](
value: Array[T],
size: Long,
classTag: ClassTag[T]) extends MemoryEntry[T] {
val memoryMode: MemoryMode = MemoryMode.ON_HEAP
}

它与SerializedMemoryEntry都是MemoryEntry[T]的子类,所有被放到同一个映射表LinkedHashMap[BlockId, MemoryEntry[_]] entries中。

另外,也存在这种可能,通过MemoryManager申请的Unroll内存大小大于该Block打开需要的内存,则会返回如下结果对象:

Left(new PartiallyUnrolledIterator( this, 
 unrollMemoryUsedByThisBlock,

unrolled = arrayValues.toIterator, rest

= Iterator.empty))

上面unrolled = arrayValues.toIterator,rest = Iterator.empty,表示在内存中可以打开迭代器中全部的数据记录,打开对象类型为DeserializedMemoryEntry[T]。

第二种,Block数据记录只能部分放到内存中:也就是说Driver或Executor上的内存有限,只可以放得下部分记录,另一部分记录内存中放不下。values记录迭代器对应的全部记录数据无法完全放在内存中,所以为了保证不发生OOM异常,首选会调用MemoryManager的acquireUnrollMemory方法去申请Unroll内存,如果可以申请到,在迭代values的过程中,需要累加计算打开(Unroll)的记录对象大小之和,使其大小不能大于申请到的Unroll内存,直到还有一部分记录无法放到申请的Unroll内存中。 最后,返回的结果对象如下所示:

Left(new PartiallyUnrolledIterator(
this, unrollMemoryUsedByThisBlock, unrolled = vector.iterator, rest = values))

上面的PartiallyUnrolledIterator中rest对应的values就是putIteratorAsValues方法传进来的迭代器参数值,该迭代器已经迭代出部分记录,放到了内存中,调用者可以继续迭代该迭代器去处理未打开(Unroll)的记录,而unrolled对应一个打开记录的迭代器。这里,PartiallyUnrolledIterator迭代器包装了vector.iterator和一个迭代出部分记录的values迭代器,调用者对PartiallyUnrolledIterator进行统一迭代能够获取到全部记录,里面包含两种类型的记录:DeserializedMemoryEntry[T]和T。

基于记录迭代器,以序列化二进制格式保存Block数据

private[storage] def putIteratorAsBytes[T](

blockId: BlockId,
values: Iterator[T],
classTag: ClassTag[T],
memoryMode: MemoryMode): Either[Partially

SerializedBlock[T], Long]

这种方式,调用这种希望将Block数据记录以二进制的格式保存在内存中。如果内存中能放得下,则返回最终的大小,否则返回一个PartiallySerializedBlock[T]迭代器。

如果Block数据记录能够完全放到内存中,则以SerializedMemoryEntry[T]格式放到内存的映射表中。如果Block数据记录只能部分放到内存中,则返回如下对象:

Left(
new PartiallySerializedBlock(
this,
serializerManager,
blockId,
serializationStream,
redirectableStream,
unrollMemoryUsedByThisBlock,
memoryMode,
bbos.toChunkedByteBuffer,
values,
classTag))

类似地,返回结果对象对调用者保持统一的迭代API视图。

DiskStore

DiskStore提供了将Block数据写入到磁盘的基本操作,它是通过DiskBlockManager来管理逻辑上Block到物理磁盘上Block文件路径的映射关系。当StorageLevel设置为如下值时,都可能会需要使用DiskStore来存储数据:

DISK_ONLY
DISK_ONLY_2
MEMORY_AND_DISK
MEMORY_AND_DISK_2
MEMORY_AND_DISK_SER
MEMORY_AND_DISK_SER_2
OFF_HEAP

DiskBlockManager管理了每个Block数据存储位置的信息,包括从Block ID到磁盘上文件的映射关系。DiskBlockManager主要有如下几个功能:

负责创建一个本地节点上的指定磁盘目录,用来存储Block数据到指定文件中

1.如果Block数据想要落盘,需要通过调用getFile方法来分配一个唯一的文件路径

2.如果想要查询一个Block是否在磁盘上,通过调用containsBlock方法来查询

3.查询当前节点上管理的全部Block文件

4.通过调用createTempLocalBlock方法,生成一个唯一Block ID,并创建一个唯一的临时文件,用来存储中间结果数据

5.通过调用createTempShuffleBlock方法,生成一个唯一Block ID,并创建一个唯一的临时文件,用来存储Shuffle过程的中间结果数据

6.DiskStore提供的基本操作接口,与MemoryStore类似,比较简单,如下所示:

通过文件流写Block数据

该种方式对应的接口方法,如下所示:

def put(blockId: BlockId)(writeFunc:
 FileOutputStream => Unit): Unit

参数指定Block ID,还有一个写Block数据到打开的文件流的函数,在调用put方法时,首先会从DiskBlockManager分配一个Block ID对应的磁盘文件路径,然后将数据写入到该文件中。

将二进制Block数据写入文件

putBytes方法实现了,将一个Buffer中的Block数据写入指定的Block ID对应的文件中,方法定义如下所示:

def putBytes(blockId: BlockId, bytes:
 ChunkedByteBuffer):
 Unit

实际上,它是调用上面的put方法,将bytes中的Block二进制数据写入到Block文件中。

从磁盘文件读取Block数据

对应方法如下所示:

def getBytes(blockId: BlockId): 
ChunkedByteBuffer

通过给定的blockId,获取磁盘上对应的Block文件的数据,以ChunkedByteBuffer的形式返回。

删除Block文件

对应的删除方法定义,如下所示:

def remove(blockId: BlockId): Boolean

通过DiskBlockManager查找到blockId对应的Block文件,然后删除掉。

BlockManager

谈到Spark中的Block数据存储,我们很容易能够想到BlockManager,他负责管理在每个Dirver和Executor上的Block数据,可能是本地或者远程的。具体操作包括查询Block、将Block保存在指定的存储中,如内存、磁盘、堆外(Off-heap)。而BlockManager依赖的后端,对Block数据进行内存、磁盘存储访问,都是基于前面讲到的MemoryStore、DiskStore。

在Spark集群中,当提交一个Application执行时,该Application对应的Driver以及所有的Executor上,都存在一个BlockManager、BlockManagerMaster,而BlockManagerMaster是负责管理各个BlockManager之间通信,这个BlockManager管理集群,如下图所示:

关于一个Application运行过程中Block的管理,主要是基于该Application所关联的一个Driver和多个Executor构建了一个Block管理集群:Driver上的(BlockManagerMaster, BlockManagerMasterEndpoint)是集群的Master角色,所有Executor上的(BlockManagerMaster, RpcEndpointRef)作为集群的Slave角色。当Executor上的Task运行时,会查询对应的RDD的某个Partition对应的Block数据是否处理过,这个过程中会触发多个BlockManager之间的通信交互。我们以ShuffleMapTask的运行为例,对应代码如下所示:

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[_], ShuffleDepen

dency[_, _, _])](
ByteBuffer.wrap(taskBinary.value), Thread.current

Thread. getContextClassLoader)
_executorDeserializeTime = System.currentTim

eMillis() - deserializeStartTime var write

r: ShuffleWriter[Any, Any] = nulltry {
val manager = SparkEnv.get.shuffleManager
writer = manager.getWriter[Any, Any](dep.shuffle

Handle, 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
}
}

一个RDD的Partition对应一个ShuffleMapTask,一个ShuffleMapTask会在一个Executor上运行,它负责处理RDD的一个Partition对应的数据,基本处理流程,如下所示:

1.根据该Partition的数据,创建一个RDDBlockId(由RDD ID和Partition Index组成),即得到一个稳定的blockId(如果该Partition数据被处理过,则可能本地或者远程Executor存储了对应的Block数据)。

2.先从BlockManager获取该blockId对应的数据是否存在,如果本地存在(已经处理过),则直接返回Block结果(BlockResult);否则,查询远程的Executor是否已经处理过该Block,处理过则直接通过网络传输到当前Executor本地,并根据StorageLevel设置,保存Block数据到本地MemoryStore或DiskStore,同时通过BlockManagerMaster上报Block数据状态(通知Driver当前的Block状态,亦即,该Block数据存储在哪个BlockManager中)。

3.如果本地及远程Executor都没有处理过该Partition对应的Block数据,则调用RDD的compute方法进行计算处理,并将处理的Block数据,根据StorageLevel设置,存储到本地MemoryStore或DiskStore。

4.根据ShuffleManager对应的ShuffleWriter,将返回的该Partition的Block数据进行Shuffle写入操作

下面,我们基于上面逻辑,详细分析在这个处理过程中重要的交互逻辑:

根据RDD获取一个Partition对应数据的记录迭代器

用户提交的Spark Application程序,会设置对应的StorageLevel,所以设置与不设置对该处理逻辑有一定影响,具有两种情况,如下图所示:

如果用户程序设置了StorageLevel,可能该Partition的数据已经处理过,那么对应的处理结果Block数据可能已经存储。一般设置的StorageLevel,或者将Block存储在内存中,或者存储在磁盘上,这里会尝试调用getOrElseUpdate()方法获取对应的Block数据,如果存在则直接返回Block对应的记录的迭代器实例,就不需要重新计算了,如果没有找到对应的已经处理过的Block数据,则调用RDD的compute()方法进行处理,处理结果根据StorageLevel设置,将Block数据存储在内存或磁盘上,缓存供后续Task重复使用。

如果用户程序没有设置StorageLevel,那么RDD对应的该Partition的数据一定没有进行处理过,即使处理过,如果没有进行Checkpointing,也需要重新计算(如果进行了Checkpointing,可以直接从缓存中获取),直接调用RDD的compute()方法进行处理。

从BlockManager查询获取Block数据

每个Executor上都有一个BlockManager实例,负责管理用户提交的该Application计算过程中产生的Block。很有可能当前Executor上存储在RDD对应Partition的经过处理后得到的Block数据,也有可能当前Executor上没有,但是其他Executor上已经处理过并缓存了Block数据,所以对应着本地获取、远程获取两种可能,本地获取交互逻辑如下图所示:

从本地获取,根据StorageLevel设置,如果是存储在内存中,则从本地的MemoryStore中查询,存在则读取并返回;如果是存储在磁盘上,则从本地的DiskStore中查询,存在则读取并返回。本地不存在,则会从远程的Executor读取,对应的组件交互逻辑,如下图所示:

远程获取交互逻辑相对比较复杂:当前Executor上的BlockManager通过BlockManagerMaster,向远程的Driver上的BlockManagerMasterEndpoint查询对应Block ID,有哪些Executor已经保存了该Block数据,Dirver返回一个包含了该Block数据的Location列表,如果对应的Location信息与当前ShuffleMapTask执行所在Executor在同一台节点上,则会优先使用该Location,因为同一节点上的多个Executor之间传输Block数据效率更高。

这里需要说明的是,如果对应的Block数据的StorageLevel设置为写磁盘,通过前面我们知道,DiskStore是通过DiskBlockManager进行管理存储到磁盘上的Block数据文件的,在同一个节点上的多个Executor共享相同的磁盘文件路径,相同的Block数据文件也就会被同一个节点上的多个Executor所共享。而对应MemoryStore,因为每个Executor对应独立的JVM实例,从而具有独立的Storage/Execution内存管理,所以使用MemoryStore不能共享同一个Block数据,但是同一个节点上的多个Executor之间的MemoryStore之间拷贝数据,比跨网络传输要高效的多。

BlockInfoManager

用户提交一个Spark Application程序,如果程序对应的DAG图相对复杂,其中很多Task计算的结果Block数据都有可能被重复使用,这种情况下如何去控制某个Executor上的Task线程去读写Block数据呢?其实,BlockInfoManager就是用来控制Block数据读写操作,并且跟踪Task读写了哪些Block数据的映射关系,这样如果两个Task都想去处理同一个RDD的同一个Partition数据,如果没有锁来控制,很可能两个Task都会计算并写同一个Block数据,从而造成混乱。我们分析每种情况下,BlockInfoManager是如何管理Block数据(同一个RDD的同一个Partition)读写的:

第一个Task请求写Block数据

这种情况下,没有其他Task写Block数据,第一个Task直接获取到写锁,并启动写Block数据到本地MemoryStore或DiskStore。如果其他写Block数据的Task也请求写锁,则该Task会阻塞,等待第一个获取写锁的Task完成写Block数据,直到第一个Task写完成,并通知其他阻塞的Task,然后其他Task需要再次获取到读锁来读取该Block数据。

第一个Task正在写Block数据,其他Task请求读Block数据

这种情况,Block数据没有完成写操作,其他读Block数据的Task只能阻塞,等待写Block的Task完成并通知读Task去读取Block数据。

Task请求读取Block数据

如果该Block数据不存在,则直接返回空,表示当前RDD的该Partition并没有被处理过。如果当前Block数据存在,并且没有其他Task在写,表示已经完成了些Block数据操作,则该Task直接读取该Block数据。

   
次浏览       
相关文章

基于EA的数据库建模
数据流建模(EA指南)
“数据湖”:概念、特征、架构与案例
在线商城数据库系统设计 思路+效果
 
相关文档

Greenplum数据库基础培训
MySQL5.1性能优化方案
某电商数据中台架构实践
MySQL高扩展架构设计
相关课程

数据治理、数据架构及数据标准
MongoDB实战课程
并发、大容量、高性能数据库设计与优化
PostgreSQL数据库实战培训