prakharjain09 commented on a change in pull request #27864: [SPARK-20732][CORE]
Decommission cache blocks to other executors when an executor is decommissioned
URL: https://github.com/apache/spark/pull/27864#discussion_r402343206
##########
File path: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
##########
@@ -1761,6 +1776,57 @@ private[spark] class BlockManager(
blocksToRemove.size
}
+ def decommissionBlockManager(): Unit = {
+ if (!blockManagerDecommissioning) {
+ logInfo("Starting block manager decommissioning process")
+ blockManagerDecommissioning = true
+ decommissionManager = Some(new BlockManagerDecommissionManager)
+ decommissionManager.foreach(_.start())
+ }
+ }
+
+ /**
+ * Tries to offload all cached RDD blocks from this BlockManager to peer
BlockManagers
+ * Visible for testing
+ */
+ def offloadRddCacheBlocks(): Unit = {
+ val replicateBlocksInfo =
master.getReplicateInfoForRDDBlocks(blockManagerId)
+
+ if (replicateBlocksInfo.nonEmpty) {
+ logInfo(s"Need to replicate ${replicateBlocksInfo.size} blocks " +
+ s"for block manager decommissioning")
+ // Refresh peer list once before starting replication
+ getPeers(true)
+ }
+
+ // Maximum number of storage replication failure which replicateBlock can
handle
+ // before giving up for one block
Review comment:
@dongjoon-hyun Renamed to
"spark.storage.decommission.maxReplicationFailuresPerBlock". Any suggestions on
2) ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
