spark存储模块之管理

操作Block的方法

MemoryManager

StorageLevel：存储级别

堆内展开地址记录：（Taskid，大小）

多线程并发访问同一个block的同步监视器

1.不断迭代读取Iterator中的数据，将数据放入追踪器vector中（周期性地检查vector中所有数据的估算大小currentSize是否已经超过了memoryThreshold。当发现currentSize超过memoryThreshold，则为当前任务请求新的保留内存）2.申请到足够多的展开内存，将数据写入内存：①将vector中的数据封装为DeserializedMemoryEntry，并重新估算vector的大小size②如果unrollMemoryUsedByThisBlock小于或等于size，说明用于展开的内存过多，需要向MemoryManager归还多余的空间。归还的内存大小为unrollMemoryUsedByThisBlock - size。之后调用 transferUnrollToStorage方法将展开Block占用的内存转换为用于存储Block的内存，此转换过程是原子的。③如果有足够的内存存储Block，则将BlockId与DeserializedMemoryEntry的映射关系放入entries并返回Right(size)④如果没有足够的内存存储Block，则创建PartiallyUnrollidIterator并返回Letf3）如果展开Iterator中所有的数据后，keepUnrolling为false，说明没有为Block申请到足够多的保留内存，此时将创建PartiallyUnrolledIterator并返回Left。

putIteratorAsValues周期性地检查是否有足够的展开内存

isShuffle：当前BlockId是否是ShuffleBlockId

classTag：存储级别

BlockManager

spark.memory.offHeap.size=maxOffHeapMemory堆外内存大小

BlockTransferService

初始化大小

_readerCount／_writerTask：锁信息

DiskBlockManager

spark.executor.memory*spark.memory.fraction（0.6）=memoryFraction=maxHeapMemory总的executor的内存*用于存储的比例=最大统一内存

isRDD：当前BlockId是否是RDDBlockId

name：Block全局唯一的标识名

提供查询blockInfo的方法提供控制BlockInfo的方法

MemoryStore

memoryManager

。。。。。

_size：大小

offHeapExecutionMemoryPool：堆外execution存储区域

doPut： 通过blockInfoManager获取操作BlockInfo的锁 调用store写入不同存储介质 以MemoryStore为例加入onHeapStorageMemoryPool加入新增大小加入MemoryStore.entries 

BlockManagerMaster

maxOffHeapMemory-offHeapStorageMemory堆外内存中：用于exetor缓存的大小

初始化

diskBlockManager

asRDDId：将当前BlockId转换为RDDBlockId。如果当前BlockId是RDDBlockId，则转换为RDDBlockId，否则返回None

BlockInfoManager

BlockID

spark.memory.storageFraction*maxOffHeapMemory=offHeapStorageMemory堆外内存用于缓存的大小

offHeapExecutionMemoryPool：堆外存储区域

MasterRef

DiskStore

onHeapStorageMemoryPool：堆内存储区域

onHeapExecutionMemoryPool：堆内execution存储区域

SparkEnv

BlockInfo

maxHeapMemory*spark.memory.storageFractionon(0.5)=HeapStorageRegionSize堆内内存中：用于存储缓存的大小

maxHeapMemory-HeapStorageRegionSize堆内内存中：用于exetor缓存的大小

Spark的存储体系：包括磁盘存储与内存存储，内存又分为堆外内存和堆内存有些数据块本身支持序列化及反序列化有些数据块还支持备份与复制