Spark源码分析之-Storage模块

Storage模块整体架构

Storage模块主要分为两层：

1. 通信层：storage模块采用的是master-slave结构来实现通信层，master和slave之间传输控制信息、状态信息，这些都是通过通信层来实现的。
2. 存储层：storage模块需要把数据存储到disk或是memory上面，有可能还需replicate到远端，这都是由存储层来实现和提供相应接口。

而其他模块若要和storage模块进行交互，storage模块提供了统一的操作类BlockManager，外部类与storage模块打交道都需要通过调用BlockManager相应接口来实现。

Storage模块通信层

首先来看一下通信层的UML类图:

其次我们来看看各个类在master和slave上所扮演的不同角色：

对于master和slave，BlockManager的创建有所不同：

• Master (client driver)

  BlockManagerMaster拥有BlockManagerMasterActor的actor和所有BlockManagerSlaveActor的ref。

• Slave (executor)

  对于slave，BlockManagerMaster则拥有BlockManagerMasterActor的ref和自身BlockManagerSlaveActor的actor。

BlockManagerMasterActor在ref和actor之间进行通信；BlockManagerSlaveActor在ref和actor之间通信。

actor和ref:

actor和ref是Akka中的两个不同的actor reference，分别由actorOf和actorFor所创建。actor类似于网络服务中的server端，它保存所有的状态信息，接收client端的请求执行并返回给客户端；ref类似于网络服务中的client端，通过向server端发起请求获取结果。

BlockManager wrap了BlockManagerMaster，通过BlockManagerMaster进行通信。Spark会在client driver和executor端创建各自的BlockManager，通过BlockManager对storage模块进行操作。

BlockManager对象在SparkEnv中被创建，创建的过程如下所示：

1. def registerOrLookup(name:String, newActor:=>Actor):ActorRef={
2. if(isDriver){
3. logInfo("Registering "+ name)
4. actorSystem.actorOf(Props(newActor), name = name)
5. }else{
6. val driverHost:String=System.getProperty("spark.driver.host","localhost")
7. val driverPort:Int=System.getProperty("spark.driver.port","7077").toInt
8. Utils.checkHost(driverHost,"Expected hostname")
9. val url ="akka://spark@%s:%s/user/%s".format(driverHost, driverPort, name)
10. logInfo("Connecting to "+ name +": "+ url)
11. actorSystem.actorFor(url)
12. }
13. }
15. val blockManagerMaster =newBlockManagerMaster(registerOrLookup(
16. "BlockManagerMaster",
17. newBlockManagerMasterActor(isLocal)))
18. val blockManager =newBlockManager(executorId, actorSystem, blockManagerMaster, serializer)

可以看到对于client driver和executor，Spark分别创建了BlockManagerMasterActor actor和ref，并被wrap到BlockManager中。

通信层传递的消息

• BlockManagerMasterActor

  • executor to client driver

    RegisterBlockManager (executor创建BlockManager以后向client driver发送请求注册自身) HeartBeat UpdateBlockInfo (更新block信息) GetPeers (请求获得其他BlockManager的id) GetLocations (获取block所在的BlockManager的id) GetLocationsMultipleBlockIds (获取一组block所在的BlockManager id)

  • client driver to client driver

    GetLocations (获取block所在的BlockManager的id) GetLocationsMultipleBlockIds (获取一组block所在的BlockManager id) RemoveExecutor (删除所保存的已经死亡的executor上的BlockManager) StopBlockManagerMaster (停止client driver上的BlockManagerMasterActor)

有些消息例如GetLocations在executor端和client driver端都会向actor请求，而其他的消息比如RegisterBlockManager只会由executor端的ref向client driver端的actor发送，于此同时例如RemoveExecutor则只会由client driver端的ref向client driver端的actor发送。

具体消息是从哪里发送，哪里接收和处理请看代码细节，在这里就不再赘述了。

• BlockManagerSlaveActor

  • client driver to executor

    RemoveBlock (删除block) RemoveRdd (删除RDD)

通信层中涉及许多控制消息和状态消息的传递以及处理，这些细节可以直接查看源码，这里就不在一一罗列。下面就只简单介绍一下exeuctor端的BlockManager是如何启动以及向client driver发送注册请求完成注册。

Register BlockManager

前面已经介绍了BlockManager对象是如何被创建出来的，当BlockManager被创建出来以后需要向client driver注册自己，下面我们来看一下这个流程：

首先BlockManager会调用initialize()初始化自己

1. privatedef initialize(){
2. master.registerBlockManager(blockManagerId, maxMemory, slaveActor)
3. ...
4. if(!BlockManager.getDisableHeartBeatsForTesting){
5. heartBeatTask = actorSystem.scheduler.schedule(0.seconds, heartBeatFrequency.milliseconds){
6. heartBeat()
7. }
8. }
9. }

在initialized()函数中首先调用BlockManagerMaster向client driver注册自己，同时设置heartbeat定时器，定时发送heartbeat报文。可以看到在注册自身的时候向client driver传递了自身的slaveActor，client driver收到slaveActor以后会将其与之对应的BlockManagerInfo存储到hash map中，以便后续通过slaveActor向executor发送命令。

BlockManagerMaster会将注册请求包装成RegisterBlockManager报文发送给client driver的BlockManagerMasterActor，BlockManagerMasterActor调用register()函数注册BlockManager：

1. privatedefregister(id:BlockManagerId, maxMemSize:Long, slaveActor:ActorRef){
2. if(id.executorId =="<driver>"&&!isLocal){
3. // Got a register message from the master node; don't register it
4. }elseif(!blockManagerInfo.contains(id)){
5. blockManagerIdByExecutor.get(id.executorId) match {
6. caseSome(manager)=>
7. // A block manager of the same executor already exists.
8. // This should never happen. Let's just quit.
9. logError("Got two different block manager registrations on "+ id.executorId)
10. System.exit(1)
11. caseNone=>
12. blockManagerIdByExecutor(id.executorId)= id
13. }
14. blockManagerInfo(id)=newBlockManagerMasterActor.BlockManagerInfo(
15. id,System.currentTimeMillis(), maxMemSize, slaveActor)
16. }
17. }

需要注意的是在client driver端也会执行上述过程，只是在最后注册的时候如果判断是"<driver>"就不进行任何操作。可以看到对应的BlockManagerInfo对象被创建并保存在hash map中。

Storage模块存储层

在RDD层面上我们了解到RDD是由不同的partition组成的，我们所进行的transformation和action是在partition上面进行的；而在storage模块内部，RDD又被视为由不同的block组成，对于RDD的存取是以block为单位进行的，本质上partition和block是等价的，只是看待的角度不同。在Spark storage模块中中存取数据的最小单位是block，所有的操作都是以block为单位进行的。

首先我们来看一下存储层的UML类图:

BlockManager对象被创建的时候会创建出MemoryStore和DiskStore对象用以存取block，同时在initialize()函数中创建BlockManagerWorker对象用以监听远程的block存取请求来进行相应处理。

1. private[storage] val memoryStore:BlockStore=newMemoryStore(this, maxMemory)
2. private[storage] val diskStore:DiskStore=
3. newDiskStore(this,System.getProperty("spark.local.dir",System.getProperty("java.io.tmpdir")))
5. privatedef initialize(){
6. ...
7. BlockManagerWorker.startBlockManagerWorker(this)
8. ...
9. }

下面就具体介绍一下对于DiskStore和MemoryStore，block的存取操作是怎样进行的。

DiskStore如何存取block

DiskStore可以配置多个folder，Spark会在不同的folder下面创建Spark文件夹，文件夹的命名方式为(spark-local-yyyyMMddHHmmss-xxxx, xxxx是一个随机数)，所有的block都会存储在所创建的folder里面。DiskStore会在对象被创建时调用createLocalDirs()来创建文件夹：

1. privatedef createLocalDirs():Array[File]={
2. logDebug("Creating local directories at root dirs '"+ rootDirs +"'")
3. val dateFormat =newSimpleDateFormat("yyyyMMddHHmmss")
4. rootDirs.split(",").map { rootDir =>
5. var foundLocalDir =false
6. var localDir:File=null
7. var localDirId:String=null
8. var tries =0
9. val rand =newRandom()
10. while(!foundLocalDir && tries < MAX_DIR_CREATION_ATTEMPTS){
11. tries +=1
12. try{
13. localDirId ="%s-%04x".format(dateFormat.format(newDate), rand.nextInt(65536))
14. localDir =newFile(rootDir,"spark-local-"+ localDirId)
15. if(!localDir.exists){
16. foundLocalDir = localDir.mkdirs()
17. }
18. }catch{
19. case e:Exception=>
20. logWarning("Attempt "+ tries +" to create local dir "+ localDir +" failed", e)
21. }
22. }
23. if(!foundLocalDir){
24. logError("Failed "+ MAX_DIR_CREATION_ATTEMPTS +
25. " attempts to create local dir in "+ rootDir)
26. System.exit(ExecutorExitCode.DISK_STORE_FAILED_TO_CREATE_DIR)
27. }
28. logInfo("Created local directory at "+ localDir)
29. localDir
30. }
31. }

在DiskStore里面，每一个block都被存储为一个file，通过计算block id的hash值将block映射到文件中，block id与文件路径的映射关系如下所示：

1. privatedef getFile(blockId:String):File={
2. logDebug("Getting file for block "+ blockId)
4. // Figure out which local directory it hashes to, and which subdirectory in that
5. val hash =Utils.nonNegativeHash(blockId)
6. val dirId = hash % localDirs.length
7. val subDirId =(hash / localDirs.length)% subDirsPerLocalDir
9. // Create the subdirectory if it doesn't already exist
10. var subDir = subDirs(dirId)(subDirId)
11. if(subDir ==null){
12. subDir = subDirs(dirId).synchronized{
13. val old = subDirs(dirId)(subDirId)
14. if(old !=null){
15. old
16. }else{
17. val newDir =newFile(localDirs(dirId),"%02x".format(subDirId))
18. newDir.mkdir()
19. subDirs(dirId)(subDirId)= newDir
20. newDir
21. }
22. }
23. }
25. newFile(subDir, blockId)
26. }

根据block id计算出hash值，将hash取模获得dirId和subDirId，在subDirs中找出相应的subDir，若没有则新建一个subDir，最后以subDir为路径、block id为文件名创建file handler，DiskStore使用此file handler将block写入文件内，代码如下所示：

1. overridedef putBytes(blockId:String, _bytes:ByteBuffer, level:StorageLevel){
2. // So that we do not modify the input offsets !
3. // duplicate does not copy buffer, so inexpensive
4. val bytes = _bytes.duplicate()
5. logDebug("Attempting to put block "+ blockId)
6. val startTime =System.currentTimeMillis
7. val file = createFile(blockId)
8. val channel =newRandomAccessFile(file,"rw").getChannel()
9. while(bytes.remaining >0){
10. channel.write(bytes)
11. }
12. channel.close()
13. val finishTime =System.currentTimeMillis
14. logDebug("Block %s stored as %s file on disk in %d ms".format(
15. blockId,Utils.bytesToString(bytes.limit),(finishTime - startTime)))
16. }

而获取block则非常简单，找到相应的文件并读取出来即可：

1. overridedef getBytes(blockId:String):Option[ByteBuffer]={
2. val file = getFile(blockId)
3. val bytes = getFileBytes(file)
4. Some(bytes)
5. }

因此在DiskStore中存取block首先是要将block id映射成相应的文件路径，接着存取文件就可以了。

MemoryStore如何存取block

相对于DiskStore需要根据block id hash计算出文件路径并将block存放到对应的文件里面，MemoryStore管理block就显得非常简单：MemoryStore内部维护了一个hash map来管理所有的block，以block id为key将block存放到hash map中。

1. caseclassEntry(value:Any, size:Long, deserialized:Boolean)
3. private val entries =newLinkedHashMap[String,Entry](32,0.75f,true)

在MemoryStore中存放block必须确保内存足够容纳下该block，若内存不足则会将block写到文件中，具体的代码如下所示：

1. overridedef putBytes(blockId:String, _bytes:ByteBuffer, level:StorageLevel){
2. // Work on a duplicate - since the original input might be used elsewhere.
3. val bytes = _bytes.duplicate()
4. bytes.rewind()
5. if(level.deserialized){
6. val values = blockManager.dataDeserialize(blockId, bytes)
7. val elements =newArrayBuffer[Any]
8. elements ++= values
9. val sizeEstimate =SizeEstimator.estimate(elements.asInstanceOf[AnyRef])
10. tryToPut(blockId, elements, sizeEstimate,true)
11. }else{
12. tryToPut(blockId, bytes, bytes.limit,false)
13. }
14. }

在tryToPut()中，首先调用ensureFreeSpace()确保空闲内存是否足以容纳block，若可以则将该block放入hash map中进行管理；若不足以容纳则通过调用dropFromMemory()将block写入文件。

1. privatedef tryToPut(blockId:String, value:Any, size:Long, deserialized:Boolean):Boolean={
2. // TODO: Its possible to optimize the locking by locking entries only when selecting blocks
3. // to be dropped. Once the to-be-dropped blocks have been selected, and lock on entries has been
4. // released, it must be ensured that those to-be-dropped blocks are not double counted for
5. // freeing up more space for another block that needs to be put. Only then the actually dropping
6. // of blocks (and writing to disk if necessary) can proceed in parallel.
7. putLock.synchronized{
8. if(ensureFreeSpace(blockId, size)){
9. val entry =newEntry(value, size, deserialized)
10. entries.synchronized{
11. entries.put(blockId, entry)
12. currentMemory += size
13. }
14. if(deserialized){
15. logInfo("Block %s stored as values to memory (estimated size %s, free %s)".format(
16. blockId,Utils.bytesToString(size),Utils.bytesToString(freeMemory)))
17. }else{
18. logInfo("Block %s stored as bytes to memory (size %s, free %s)".format(
19. blockId,Utils.bytesToString(size),Utils.bytesToString(freeMemory)))
20. }
21. true
22. }else{
23. // Tell the block manager that we couldn't put it in memory so that it can drop it to
24. // disk if the block allows disk storage.
25. val data =if(deserialized){
26. Left(value.asInstanceOf[ArrayBuffer[Any]])
27. }else{
28. Right(value.asInstanceOf[ByteBuffer].duplicate())
29. }
30. blockManager.dropFromMemory(blockId, data)
31. false
32. }
33. }
34. }

而从MemoryStore中取得block则非常简单，只需从hash map中取出block id对应的value即可。

1. overridedef getValues(blockId:String):Option[Iterator[Any]]={
2. val entry = entries.synchronized{
3. entries.get(blockId)
4. }
5. if(entry ==null){
6. None
7. }elseif(entry.deserialized){
8. Some(entry.value.asInstanceOf[ArrayBuffer[Any]].iterator)
9. }else{
10. val buffer = entry.value.asInstanceOf[ByteBuffer].duplicate()// Doesn't actually copy data
11. Some(blockManager.dataDeserialize(blockId, buffer))
12. }
13. }

Put or Get block through BlockManager

上面介绍了DiskStore和MemoryStore对于block的存取操作，那么我们是要直接与它们交互存取数据吗，还是封装了更抽象的接口使我们无需关心底层？

BlockManager为我们提供了put()和get()函数，用户可以使用这两个函数对block进行存取而无需关心底层实现。

首先我们来看一下put()函数的实现：

1. def put(blockId:String, values:ArrayBuffer[Any], level:StorageLevel,
2. tellMaster:Boolean=true):Long={
4. ...
6. // Remember the block's storage level so that we can correctly drop it to disk if it needs
7. // to be dropped right after it got put into memory. Note, however, that other threads will
8. // not be able to get() this block until we call markReady on its BlockInfo.
9. val myInfo ={
10. val tinfo =newBlockInfo(level, tellMaster)
11. // Do atomically !
12. val oldBlockOpt = blockInfo.putIfAbsent(blockId, tinfo)
14. if(oldBlockOpt.isDefined){
15. if(oldBlockOpt.get.waitForReady()){
16. logWarning("Block "+ blockId +" already exists on this machine; not re-adding it")
17. return oldBlockOpt.get.size
18. }
20. // TODO: So the block info exists - but previous attempt to load it (?) failed. What do we do now ? Retry on it ?
21. oldBlockOpt.get
22. }else{
23. tinfo
24. }
25. }
27. val startTimeMs =System.currentTimeMillis
29. // If we need to replicate the data, we'll want access to the values, but because our
30. // put will read the whole iterator, there will be no values left. For the case where
31. // the put serializes data, we'll remember the bytes, above; but for the case where it
32. // doesn't, such as deserialized storage, let's rely on the put returning an Iterator.
33. var valuesAfterPut:Iterator[Any]=null
35. // Ditto for the bytes after the put
36. var bytesAfterPut:ByteBuffer=null
38. // Size of the block in bytes (to return to caller)
39. var size =0L
41. myInfo.synchronized{
42. logTrace("Put for block "+ blockId +" took "+Utils.getUsedTimeMs(startTimeMs)
43. +" to get into synchronized block")
45. var marked =false
46. try{
47. if(level.useMemory){
48. // Save it just to memory first, even if it also has useDisk set to true; we will later
49. // drop it to disk if the memory store can't hold it.
50. val res = memoryStore.putValues(blockId, values, level,true)
51. size = res.size
52. res.data match {
53. caseRight(newBytes)=> bytesAfterPut = newBytes
54. caseLeft(newIterator)=> valuesAfterPut = newIterator
55. }
56. }else{
57. // Save directly to disk.
58. // Don't get back the bytes unless we replicate them.
59. val askForBytes = level.replication >1
60. val res = diskStore.putValues(blockId, values, level, askForBytes)
61. size = res.size
62. res.data match {
63. caseRight(newBytes)=> bytesAfterPut = newBytes
64. case _ =>
65. }
66. }
68. // Now that the block is in either the memory or disk store, let other threads read it,
69. // and tell the master about it.
70. marked =true
71. myInfo.markReady(size)
72. if(tellMaster){
73. reportBlockStatus(blockId, myInfo)
74. }
75. }finally{
76. // If we failed at putting the block to memory/disk, notify other possible readers
77. // that it has failed, and then remove it from the block info map.
78. if(! marked){
79. // Note that the remove must happen before markFailure otherwise another thread
80. // could've inserted a new BlockInfo before we remove it.
81. blockInfo.remove(blockId)
82. myInfo.markFailure()
83. logWarning("Putting block "+ blockId +" failed")
84. }
85. }
86. }
87. logDebug("Put block "+ blockId +" locally took "+Utils.getUsedTimeMs(startTimeMs))
89. // Replicate block if required
90. if(level.replication >1){
91. val remoteStartTime =System.currentTimeMillis
92. // Serialize the block if not already done
93. if(bytesAfterPut ==null){
94. if(valuesAfterPut ==null){
95. thrownewSparkException(
96. "Underlying put returned neither an Iterator nor bytes! This shouldn't happen.")
97. }
98. bytesAfterPut = dataSerialize(blockId, valuesAfterPut)
99. }
100. replicate(blockId, bytesAfterPut, level)
101. logDebug("Put block "+ blockId +" remotely took "+Utils.getUsedTimeMs(remoteStartTime))
102. }
103. BlockManager.dispose(bytesAfterPut)
105. return size
106. }

对于put()操作，主要分为以下3个步骤：

1. 为block创建BlockInfo结构体存储block相关信息，同时将其加锁使其不能被访问。
2. 根据block的storage level将block存储到memory或是disk上，同时解锁标识该block已经ready，可被访问。
3. 根据block的replication数决定是否将该block replicate到远端。

接着我们来看一下get()函数的实现：

1. defget(blockId:String):Option[Iterator[Any]]={
2. val local= getLocal(blockId)
3. if(local.isDefined){
4. logInfo("Found block %s locally".format(blockId))
5. returnlocal
6. }
7. val remote = getRemote(blockId)
8. if(remote.isDefined){
9. logInfo("Found block %s remotely".format(blockId))
10. return remote
11. }
12. None
13. }

get()首先会从local的BlockManager中查找block，如果找到则返回相应的block，若local没有找到该block，则发起请求从其他的executor上的BlockManager中查找block。在通常情况下Spark任务的分配是根据block的分布决定的，任务往往会被分配到拥有block的节点上，因此getLocal()就能找到所需的block；但是在资源有限的情况下，Spark会将任务调度到与block不同的节点上，这样就必须通过getRemote()来获得block。

我们先来看一下getLocal():

1. def getLocal(blockId:String):Option[Iterator[Any]]={
2. logDebug("Getting local block "+ blockId)
3. val info = blockInfo.get(blockId).orNull
4. if(info !=null){
5. info.synchronized{
7. // In the another thread is writing the block, wait for it to become ready.
8. if(!info.waitForReady()){
9. // If we get here, the block write failed.
10. logWarning("Block "+ blockId +" was marked as failure.")
11. returnNone
12. }
14. val level = info.level
15. logDebug("Level for block "+ blockId +" is "+ level)
17. // Look for the block in memory
18. if(level.useMemory){
19. logDebug("Getting block "+ blockId +" from memory")
20. memoryStore.getValues(blockId) match {
21. caseSome(iterator)=>
22. returnSome(iterator)
23. caseNone=>
24. logDebug("Block "+ blockId +" not found in memory")
25. }
26. }
28. // Look for block on disk, potentially loading it back into memory if required
29. if(level.useDisk){
30. logDebug("Getting block "+ blockId +" from disk")
31. if(level.useMemory && level.deserialized){
32. diskStore.getValues(blockId) match {
33. caseSome(iterator)=>
34. // Put the block back in memory before returning it
35. // TODO: Consider creating a putValues that also takes in a iterator ?
36. val elements =newArrayBuffer[Any]
37. elements ++= iterator
38. memoryStore.putValues(blockId, elements, level,true).data match {
39. caseLeft(iterator2)=>
40. returnSome(iterator2)
41. case _ =>
42. thrownewException("Memory store did not return back an iterator")
43. }
44. caseNone=>
45. thrownewException("Block "+ blockId +" not found on disk, though it should be")
46. }
47. }elseif(level.useMemory &&!level.deserialized){
48. // Read it as a byte buffer into memory first, then return it
49. diskStore.getBytes(blockId) match {
50. caseSome(bytes)=>
51. // Put a copy of the block back in memory before returning it. Note that we can't
52. // put the ByteBuffer returned by the disk store as that's a memory-mapped file.
53. // The use of rewind assumes this.
54. assert(0== bytes.position())
55. val copyForMemory =ByteBuffer.allocate(bytes.limit)
56. copyForMemory.put(bytes)
57. memoryStore.putBytes(blockId, copyForMemory, level)
58. bytes.rewind()
59. returnSome(dataDeserialize(blockId, bytes))
60. caseNone=>
61. thrownewException("Block "+ blockId +" not found on disk, though it should be")
62. }
63. }else{
64. diskStore.getValues(blockId) match {
65. caseSome(iterator)=>
66. returnSome(iterator)
67. caseNone=>
68. thrownewException("Block "+ blockId +" not found on disk, though it should be")
69. }
70. }
71. }
72. }
73. }else{
74. logDebug("Block "+ blockId +" not registered locally")
75. }
76. returnNone
77. }

getLocal()首先会根据block id获得相应的BlockInfo并从中取出该block的storage level，根据storage level的不同getLocal()又进入以下不同分支：

1. level.useMemory == true：从memory中取出block并返回，若没有取到则进入分支2。
2. level.useDisk == true:
   • level.useMemory == true: 将block从disk中读出并写入内存以便下次使用时直接从内存中获得，同时返回该block。
   • level.useMemory == false: 将block从disk中读出并返回
3. level.useDisk == false: 没有在本地找到block，返回None。

接下来我们来看一下getRemote()：

1. def getRemote(blockId:String):Option[Iterator[Any]]={
2. if(blockId ==null){
3. thrownewIllegalArgumentException("Block Id is null")
4. }
5. logDebug("Getting remote block "+ blockId)
6. // Get locations of block
7. val locations = master.getLocations(blockId)
9. // Get block from remote locations
10. for(loc <- locations){
11. logDebug("Getting remote block "+ blockId +" from "+ loc)
12. val data =BlockManagerWorker.syncGetBlock(
13. GetBlock(blockId),ConnectionManagerId(loc.host, loc.port))
14. if(data !=null){
15. returnSome(dataDeserialize(blockId, data))
16. }
17. logDebug("The value of block "+ blockId +" is null")
18. }
19. logDebug("Block "+ blockId +" not found")
20. returnNone
21. }

getRemote()首先取得该block的所有location信息，然后根据location向远端发送请求获取block，只要有一个远端返回block该函数就返回而不继续发送请求。

至此我们简单介绍了BlockManager类中的get()和put()函数，使用这两个函数外部类可以轻易地存取block数据。

Partition如何转化为Block

在storage模块里面所有的操作都是和block相关的，但是在RDD里面所有的运算都是基于partition的，那么partition是如何与block对应上的呢？

RDD计算的核心函数是iterator()函数：

1. finaldef iterator(split:Partition, context:TaskContext):Iterator[T]={
2. if(storageLevel !=StorageLevel.NONE){
3. SparkEnv.get.cacheManager.getOrCompute(this, split, context, storageLevel)
4. }else{
5. computeOrReadCheckpoint(split, context)
6. }
7. }

如果当前RDD的storage level不是NONE的话，表示该RDD在BlockManager中有存储，那么调用CacheManager中的getOrCompute()函数计算RDD，在这个函数中partition和block发生了关系：

首先根据RDD id和partition index构造出block id (rdd_xx_xx)，接着从BlockManager中取出相应的block。

• 如果该block存在，表示此RDD在之前已经被计算过和存储在BlockManager中，因此取出即可，无需再重新计算。
• 如果该block不存在则需要调用RDD的computeOrReadCheckpoint()函数计算出新的block，并将其存储到BlockManager中。

需要注意的是block的计算和存储是阻塞的，若另一线程也需要用到此block则需等到该线程block的loading结束。

1. def getOrCompute[T](rdd: RDD[T], split:Partition, context:TaskContext, storageLevel:StorageLevel)
2. :Iterator[T]={
3. val key ="rdd_%d_%d".format(rdd.id, split.index)
4. logDebug("Looking for partition "+ key)
5. blockManager.get(key) match {
6. caseSome(values)=>
7. // Partition is already materialized, so just return its values
8. return values.asInstanceOf[Iterator[T]]
10. caseNone=>
11. // Mark the split as loading (unless someone else marks it first)
12. loading.synchronized{
13. if(loading.contains(key)){
14. logInfo("Another thread is loading %s, waiting for it to finish...".format (key))
15. while(loading.contains(key)){
16. try{loading.wait()}catch{case _ :Throwable=>}
17. }
18. logInfo("Finished waiting for %s".format(key))
19. // See whether someone else has successfully loaded it. The main way this would fail
20. // is for the RDD-level cache eviction policy if someone else has loaded the same RDD
21. // partition but we didn't want to make space for it. However, that case is unlikely
22. // because it's unlikely that two threads would work on the same RDD partition. One
23. // downside of the current code is that threads wait serially if this does happen.
24. blockManager.get(key) match {
25. caseSome(values)=>
26. return values.asInstanceOf[Iterator[T]]
27. caseNone=>
28. logInfo("Whoever was loading %s failed; we'll try it ourselves".format (key))
29. loading.add(key)
30. }
31. }else{
32. loading.add(key)
33. }
34. }
35. try{
36. // If we got here, we have to load the split
37. logInfo("Partition %s not found, computing it".format(key))
38. val computedValues = rdd.computeOrReadCheckpoint(split, context)
39. // Persist the result, so long as the task is not running locally
40. if(context.runningLocally){return computedValues }
41. val elements =newArrayBuffer[Any]
42. elements ++= computedValues
43. blockManager.put(key, elements, storageLevel,true)
44. return elements.iterator.asInstanceOf[Iterator[T]]
45. }finally{
46. loading.synchronized{
47. loading.remove(key)
48. loading.notifyAll()
49. }
50. }
51. }
52. }

这样RDD的transformation、action就和block数据建立了联系，虽然抽象上我们的操作是在partition层面上进行的，但是partition最终还是被映射成为block，因此实际上我们的所有操作都是对block的处理和存取。

 

http://jerryshao.me/architecture/2013/10/08/spark-storage-module-analysis/