[WIP][SPARK-25888][Core] Serve local disk persisted blocks by the external service after releasing executor by dynamic allocation

[attilapiros]

What changes were proposed in this pull request?

Problem statement

An executor which has persisted blocks does not consider to be idle and this way ready to be released by dynamic allocation after the regular timeout spark.dynamicAllocation.executorIdleTimeout but there is separate configuration spark.dynamicAllocation.cachedExecutorIdleTimeout which defaults to Integer.MAX_VALUE. This is because releasing the executor also means losing the persisted blocks (as the metadata for individual blocks called BlockInfo are kept in memory) and when the RDD is referenced latter on this lost blocks will be recomputed.
On the other hand keeping the executors too long without any task to work on is also a waste of resources (as executors are reserved for the application by the resource manager).

Solution

This PR focuses on the first part of SPARK-25888: it extends the external shuffle service with the capability to serve RDD blocks which are persisted on the local disk store by the executors. Moreover when a block is reported to be persisted on to disk the external shuffle service instance running on the same host as the executor is also registered (along with the reporting block manager) as a possible location for fetching it.

Some implementation detail

Some explanation about the decisions made during the development:

• Introducing BlockTransferClient and BlockTransferClientSync instead of moving BlockTransferService#fetchBlockSync() method to ShuffleClient:
  ShuffleClient is in the spark-network-shuffle artifact where neither EncryptedManagedBuffer and SparkException (used in the awaitResult() which should have been reimplemented here with a Java promise) are not available.
• the location list to fetch a block was randomized but the groups (same host, same rack, others) order was kept. In this PR the order of groups are kept and external shuffle service added to the end of the each group.
• BlockManagerInfo is not introduced for external shuffle service but only a lightweight solution is taken. A hash map from BlockId to BlockStatus is introduced. A type alias would make the source more readable but I know it is discouraged. On the other hand a new class wrapping this hash map would introduce unnecessary indirection.

How was this patch tested?

Unit tests

ExternalShuffleServiceSuite

Here the complete use case is tested by the "SPARK-25888: using external shuffle service fetching disk persisted blocks" with a tiny difference: here the executor is killed manually, this way the test is a bit faster than waiting for the idle timeout.

ExternalShuffleBlockHandlerSuite

Tests the fetching of the RDD blocks via the external shuffle service.

BlockManagerInfoSuite

This a new suite. As the BlockManagerInfo behaviour depends very much on whether the external shuffle service enabled or not all the tests are executed with and without it.

BlockManagerSuite

Tests the sorting of the block locations.

Manually on YARN

Spark App was:

package com.mycompany

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.storage.StorageLevel

object TestAppDiskOnlyLevel {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("test-app")

    println("Attila: START")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(0 until 100, 10)
      .map { i =>
        println(s"Attila: calculate first rdd i=$i")
        Thread.sleep(1000)
        i
      }

    rdd.persist(StorageLevel.DISK_ONLY)
    rdd.count()

    println("Attila: First RDD is processed, waiting for 60 sec")

    Thread.sleep(60 * 1000)

    println("Attila: Num executors must be 0 as executorIdleTimeout is way over")

    val rdd2 = sc.parallelize(0 until 10, 1)
      .map(i => (i, 1))
      .persist(StorageLevel.DISK_ONLY)

    rdd2.count()

    println("Attila: Second RDD with one partition (only one executors must be alive)")
    println("Attila: Calling collect on the first RDD:")

    rdd.collect()

    println("Attila: STOP")
  }
}

I have submitted with the following configuration:

spark-submit --master yarn \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.executorIdleTimeout=30 \
  --conf spark.dynamicAllocation.cachedExecutorIdleTimeout=90 \
  --class com.mycompany.TestAppDiskOnlyLevel dyn_alloc_demo-core_2.11-0.1.0-SNAPSHOT-jar-with-dependencies.jar

Checked the result by filtering for the side effect of the task calculations:

[user@server ~]$ yarn logs -applicationId application_1556299359453_0001 | grep "Attila: calculate" | wc -l
WARNING: YARN_OPTS has been replaced by HADOOP_OPTS. Using value of YARN_OPTS.
19/04/26 10:31:59 INFO client.RMProxy: Connecting to ResourceManager at apiros-1.gce.company.com/172.31.115.165:8032
100

So it is only 100 task execution and not 200 (which would be the case for re-computation).

Moreover from the submit/launcher log we can see executors really stopped in between (see the new total is 0 before the last line):

[user@server ~]$ grep "Attila: Num executors must be 0" -B 2 spark-submit.log
19/04/26 10:24:27 INFO cluster.YarnScheduler: Executor 9 on apiros-3.gce.company.com killed by driver.
19/04/26 10:24:27 INFO spark.ExecutorAllocationManager: Existing executor 9 has been removed (new total is 0)
Attila: Num executors must be 0 as executorIdleTimeout is way over

Full spark submit log

I have done a test also after changing the DISK_ONLY storage level to MEMORY_ONLY for the first RDD. After this change during the 60sec waiting no executor was removed.

[squito]

have you done any tests on what happens when a block is cached as MEMORY_AND_DISK? I know that the meaning of storageLevel is a little ambiguous and can change depending on where the msg is. We should make sure you're not getting useDisk = true here if the block is only in memory currently.

[attilapiros]

I think according to the code we are safe here as a few lines above we are calling updateBlockInfo with the storageLevel:

spark/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala

Lines 435 to 437 in a25e9aa

	blockManagerInfo(blockManagerId).updateBlockInfo(blockId, storageLevel, memSize, diskSize)

And the same storageLevel is used to create the BlockStatus where only the diskSize is filled:

spark/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala

Lines 610 to 612 in a25e9aa

	if (storageLevel.useDisk) {
	blockStatus = BlockStatus(storageLevel, memSize = 0, diskSize = diskSize)
	_blocks.put(blockId, blockStatus)

[attilapiros]

A bit weird that in the map we are overwriting the one which calculated for (useMemory=true):

spark/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala

Lines 596 to 598 in a25e9aa

	blockStatus = BlockStatus(storageLevel, memSize = memSize, diskSize = 0)
	_blocks.put(blockId, blockStatus)
	_remainingMem -= memSize

But this cleanups I plan to do that in separate PR to keep my current PR focused to the new feature and make the reviewers life easier.

[attilapiros]

I checked how the the message is created and this is where the storageLevel is computed (called from doPutIterator):

spark/core/src/main/scala/org/apache/spark/storage/BlockManager.scala

Lines 693 to 697 in a25e9aa

	val onDisk = level.useDisk && diskStore.contains(blockId)
	val deserialized = if (inMem) level.deserialized else false
	val replication = if (inMem || onDisk) level.replication else 1
	val storageLevel = StorageLevel(
	useDisk = onDisk,

First line and last line of the permalink.