spark 存储机制详解

 1   private val cacheLocs = new HashMap[Int, Array[Seq[TaskLocation]]]
 2   
 3   private def getCacheLocs(rdd: RDD[_]): Array[Seq[TaskLocation]] = {
 4     if (!cacheLocs.contains(rdd.id)) {
 5       val blockIds = rdd.partitions.indices.map(index => RDDBlockId(rdd.id, index)).toArray[BlockId]
 6       val locs = BlockManager.blockIdsToBlockManagers(blockIds, env, blockManagerMaster)
 7       cacheLocs(rdd.id) = blockIds.map { id =>
 8         locs.getOrElse(id, Nil).map(bm => TaskLocation(bm.host, bm.executorId))
 9       }
10     }
11     cacheLocs(rdd.id)
12   }

 1   def blockIdsToBlockManagers(
 2       blockIds: Array[BlockId],
 3       env: SparkEnv,
 4       blockManagerMaster: BlockManagerMaster = null): Map[BlockId, Seq[BlockManagerId]] = {
 5 
 6     // blockManagerMaster != null is used in tests
 7     assert(env != null || blockManagerMaster != null)
 8     val blockLocations: Seq[Seq[BlockManagerId]] = if (blockManagerMaster == null) {
 9       env.blockManager.getLocationBlockIds(blockIds)
10     } else {
11       blockManagerMaster.getLocations(blockIds)
12     }
13 
14     val blockManagers = new HashMap[BlockId, Seq[BlockManagerId]]
15     for (i <- 0 until blockIds.length) {
16       blockManagers(blockIds(i)) = blockLocations(i)
17     }
18     blockManagers.toMap
19   }

1   def getLocations(blockId: BlockId): Seq[BlockManagerId] = {
2     askDriverWithReply[Seq[BlockManagerId]](GetLocations(blockId))
3   }
4   private def askDriverWithReply[T](message: Any): T = {
5     AkkaUtils.askWithReply(message, driverActor, AKKA_RETRY_ATTEMPTS, AKKA_RETRY_INTERVAL_MS,
6       timeout)
7   }

1   override def receiveWithLogging = {
2     case GetLocations(blockId) =>
3       sender ! getLocations(blockId)
4     case ... =>
5   }
6 
7   private def getLocations(blockId: BlockId): Seq[BlockManagerId] = {
8     if (blockLocations.containsKey(blockId)) blockLocations.get(blockId).toSeq else Seq.empty
9   }

1   def removeBlock(blockId: BlockId) {
2     askDriverWithReply(RemoveBlock(blockId))
3   }
4 
5   private def askDriverWithReply[T](message: Any): T = {
6     AkkaUtils.askWithReply(message, driverActor, AKKA_RETRY_ATTEMPTS, AKKA_RETRY_INTERVAL_MS,
7       timeout)
8   }

 1   override def receiveWithLogging = {
 2     case RemoveBlock(blockId) =>
 3       removeBlockFromWorkers(blockId)
 4       sender ! true
 5     case ... =>
 6   }
 7 
 8   private def removeBlockFromWorkers(blockId: BlockId) {
 9     val locations = blockLocations.get(blockId)
10     if (locations != null) {
11       locations.foreach { blockManagerId: BlockManagerId =>
12         val blockManager = blockManagerInfo.get(blockManagerId)
13         if (blockManager.isDefined) {
14           // Remove the block from the slave‘s BlockManager.
15           // Doesn‘t actually wait for a confirmation and the message might get lost.
16           // If message loss becomes frequent, we should add retry logic here.
17           blockManager.get.slaveActor.ask(RemoveBlock(blockId))(akkaTimeout)
18         }
19       }
20     }
21   }

1   override def receiveWithLogging = {
2     case RemoveBlock(blockId) =>
3       doAsync[Boolean]("removing block " + blockId, sender) {
4         blockManager.removeBlock(blockId)
5         true
6       }
7     case ... =>
8   }

 1   def removeBlock(blockId: BlockId, tellMaster: Boolean = true): Unit = {
 2     logInfo(s"Removing block $blockId")
 3     val info = blockInfo.get(blockId).orNull
 4     if (info != null) {
 5       info.synchronized {
 6         // Removals are idempotent in disk store and memory store. At worst, we get a warning.
 7         val removedFromMemory = memoryStore.remove(blockId)
 8         val removedFromDisk = diskStore.remove(blockId)
 9         val removedFromTachyon = if (tachyonInitialized) tachyonStore.remove(blockId) else false
10         if (!removedFromMemory && !removedFromDisk && !removedFromTachyon) {
11           logWarning(s"Block $blockId could not be removed as it was not found in either " +
12             "the disk, memory, or tachyon store")
13         }
14         blockInfo.remove(blockId)
15         if (tellMaster && info.tellMaster) {
16           val status = getCurrentBlockStatus(blockId, info)
17           reportBlockStatus(blockId, info, status)
18         }
19       }
20     } else {
21       // The block has already been removed; do nothing.
22       logWarning(s"Asked to remove block $blockId, which does not exist")
23     }
24   }