KE-17158 renew executor in standalone (apache#171)

* dynamic request executors and renew mem/core in standalone
microbearz · Sep 15, 2020 · b735acc · b735acc
1 parent c43c495
commit b735acc
Show file tree

Hide file tree

Showing 48 changed files with 206 additions and 42 deletions.
diff --git a/assembly/pom.xml b/assembly/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/common/kvstore/pom.xml b/common/kvstore/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/common/network-common/pom.xml b/common/network-common/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/common/network-shuffle/pom.xml b/common/network-shuffle/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/common/network-yarn/pom.xml b/common/network-yarn/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/common/sketch/pom.xml b/common/sketch/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/common/tags/pom.xml b/common/tags/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/common/unsafe/pom.xml b/common/unsafe/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/core/pom.xml b/core/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala
@@ -46,6 +46,14 @@ private[spark] trait ExecutorAllocationClient {
       localityAwareTasks: Int,
       hostToLocalTaskCount: Map[String, Int]): Boolean
 
+  private[spark] def requestTotalExecutors(
+                                            numExecutors: Int,
+                                            forceKillOldExecutors: Boolean,
+                                            newMemoryPerExecutorMB: Option[Int],
+                                            newCoresPerExecutor: Option[Int]): Boolean = {
+    requestTotalExecutors(numExecutors, 0, Map.empty)
+  }
+
   /**
    * Request an additional number of executors from the cluster manager.
    * @return whether the request is acknowledged by the cluster manager.

diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -1643,6 +1643,23 @@ class SparkContext(config: SparkConf) extends Logging {
     }
   }
 
+  @DeveloperApi
+  def requestTotalExecutors(
+                         numExecutors: Int,
+                         forceKillOldExecutors: Boolean,
+                         newMemoryPerExecutorMB: Option[Int],
+                         newCoresPerExecutor: Option[Int]
+                                  ): Boolean = {
+    schedulerBackend match {
+      case b: ExecutorAllocationClient =>
+        b.requestTotalExecutors(numExecutors, forceKillOldExecutors,
+          newMemoryPerExecutorMB, newCoresPerExecutor)
+      case _ =>
+        logWarning("Requesting executors is not supported by current scheduler.")
+        false
+    }
+  }
+
   /**
    * :: DeveloperApi ::
    * Request an additional number of executors from the cluster manager.

diff --git a/core/src/main/scala/org/apache/spark/deploy/ApplicationDescription.scala b/core/src/main/scala/org/apache/spark/deploy/ApplicationDescription.scala
@@ -22,13 +22,13 @@ import java.net.URI
 private[spark] case class ApplicationDescription(
     name: String,
     maxCores: Option[Int],
-    memoryPerExecutorMB: Int,
+    var memoryPerExecutorMB: Int,
     command: Command,
     appUiUrl: String,
     eventLogDir: Option[URI] = None,
     // short name of compression codec used when writing event logs, if any (e.g. lzf)
     eventLogCodec: Option[String] = None,
-    coresPerExecutor: Option[Int] = None,
+    var coresPerExecutor: Option[Int] = None,
     // number of executors this application wants to start with,
     // only used if dynamic allocation is enabled
     initialExecutorLimit: Option[Int] = None,

diff --git a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
@@ -142,6 +142,11 @@ object DeployMessages {
 
   case class RequestExecutors(appId: String, requestedTotal: Int)
 
+  case class RequestExecutorsRenew(appId: String, requestedTotal: Int
+                                            , forceKillOldExecutors: Boolean
+                                            , newMemoryPerExecutorMB: Option[Int]
+                                            , newCoresPerExecutor: Option[Int])
+
   case class KillExecutors(appId: String, executorIds: Seq[String])
 
   // Master to AppClient

diff --git a/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala b/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala
@@ -207,6 +207,14 @@ private[spark] class StandaloneAppClient(
             logWarning("Attempted to request executors before registering with Master.")
             context.reply(false)
         }
+      case r: RequestExecutorsRenew =>
+        logInfo("send RequestExecutorsRenew rpc")
+        master match {
+          case Some(m) => askAndReplyAsync(m, context, r)
+          case None =>
+            logWarning("Attempted to request and renew executors.")
+            context.reply(false)
+        }
 
       case k: KillExecutors =>
         master match {
@@ -304,6 +312,21 @@ private[spark] class StandaloneAppClient(
     }
   }
 
+  def requestTotalExecutors(requestedTotal: Int,
+                                   forceKillOldExecutors: Boolean,
+                                   newMemoryPerExecutorMB: Option[Int],
+                                   newCoresPerExecutor: Option[Int]): Future[Boolean] = {
+    if (endpoint.get != null && appId.get != null) {
+      endpoint.get.ask[Boolean](
+        RequestExecutorsRenew(appId.get,
+          requestedTotal, forceKillOldExecutors,
+          newMemoryPerExecutorMB, newCoresPerExecutor))
+    } else {
+      logWarning("Attempted to request and refresh executors before driver fully initialized.")
+      Future.successful(false)
+    }
+  }
+
   /**
    * Kill the given list of executors through the Master.
    * @return whether the kill request is acknowledged.

diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
@@ -503,6 +503,11 @@ private[deploy] class Master(
     case RequestExecutors(appId, requestedTotal) =>
       context.reply(handleRequestExecutors(appId, requestedTotal))
 
+    case RequestExecutorsRenew(appId, requestedTotal, forceKillOldExecutors,
+    newMemoryPerExecutorMB, newCoresPerExecutor) =>
+      context.reply(handleRefreshApplicationAndExecutors(appId, requestedTotal,
+        forceKillOldExecutors, newMemoryPerExecutorMB, newCoresPerExecutor))
+
     case KillExecutors(appId, executorIds) =>
       val formattedExecutorIds = formatExecutorIds(executorIds)
       context.reply(handleKillExecutors(appId, formattedExecutorIds))
@@ -921,6 +926,66 @@ private[deploy] class Master(
     }
   }
 
+  private def printExecutorStatus(appInfo: ApplicationInfo): Unit = {
+    logInfo(s"${appInfo.id} current executor:" +
+      appInfo.executors.values.map(executor => s"(${executor.id},${executor.state})")
+        .toList.mkString(" "))
+  }
+
+  private def handleRefreshApplicationAndExecutors(appId: String, requestedTotal: Int
+                                                   , forceKillOldExecutors: Boolean
+                                                   , newMemoryPerExecutorMB: Option[Int]
+                                                   , newCoresPerExecutor: Option[Int]): Boolean = {
+    idToApp.get(appId) match {
+      case Some(appInfo) =>
+        val appDesc = appInfo.desc
+        logInfo(s"Application $appId before requested and " +
+          s"renew to set total executors to $requestedTotal, " +
+          s"newMemoryPerExecutorMB to ${appDesc.memoryPerExecutorMB}, " +
+          s"newCoresPerExecutor to ${appDesc.coresPerExecutor}, ")
+        printExecutorStatus(appInfo)
+
+        newMemoryPerExecutorMB match {
+          case Some(i) => appDesc.memoryPerExecutorMB = i
+        }
+
+        if (newCoresPerExecutor != null) {
+          appDesc.coresPerExecutor = newCoresPerExecutor
+        }
+        appInfo.executorLimit = requestedTotal
+
+        handleKillOldExecutors(appInfo, forceKillOldExecutors)
+        schedule()
+
+        logInfo(s"Application $appId after requested and renew to set " +
+          s"newMemoryPerExecutorMB to ${appDesc.memoryPerExecutorMB}, " +
+          s"newCoresPerExecutor to ${appDesc.coresPerExecutor}, ")
+        printExecutorStatus(appInfo)
+        true
+      case None =>
+        logWarning(s"Unknown application $appId requested $requestedTotal total executors.")
+        false
+    }
+  }
+
+  private def handleKillOldExecutors(appInfo: ApplicationInfo,
+                                     forceKillOldExecutors: Boolean): Boolean = {
+    for (execId <- appInfo.executors.keys) {
+      appInfo.executors.get(execId) match {
+        case Some(executorDesc) =>
+          if (forceKillOldExecutors || ExecutorState.isFinished(executorDesc.state)) {
+            appInfo.removeExecutor(executorDesc)
+            killExecutor(executorDesc)
+            logInfo("remote kill executor " + executorDesc.fullId +
+              " on worker " + executorDesc.worker.id)
+          }
+        case None =>
+          logWarning("executor not found when kill")
+      }
+    }
+    true
+  }
+
   /**
    * Handle a kill request from the given application.
    *

diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -588,6 +588,32 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
     defaultAskTimeout.awaitResult(response)
   }
 
+  final override def requestTotalExecutors(numExecutors: Int,
+                                            forceKillOldExecutors: Boolean,
+                                            newMemoryPerExecutorMB: Option[Int],
+                                            newCoresPerExecutor: Option[Int]): Boolean = {
+    if (numExecutors < 0) {
+      throw new IllegalArgumentException(
+        "Attempted to request a negative number of executor(s) " +
+          s"$numExecutors from the cluster manager. Please specify a positive number!")
+    }
+    logInfo(s"request to total executor and renew $numExecutors , " +
+      s"forceKillOldExecutors:$forceKillOldExecutors , " +
+      s"newMemoryPerExecutorMB:$newMemoryPerExecutorMB ," +
+      s"newCoresPerExecutor:$newCoresPerExecutor")
+
+    val response = synchronized {
+      this.requestedTotalExecutors = numExecutors
+      numPendingExecutors =
+        math.max(numExecutors - numExistingExecutors + executorsPendingToRemove.size, 0)
+
+      doRequestTotalExecutors(numExecutors,
+        forceKillOldExecutors, newMemoryPerExecutorMB, newCoresPerExecutor)
+    }
+
+    defaultAskTimeout.awaitResult(response)
+  }
+
   /**
    * Request executors from the cluster manager by specifying the total number desired,
    * including existing pending and running executors.
@@ -603,6 +629,11 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   protected def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] =
     Future.successful(false)
 
+  protected def doRequestTotalExecutors(requestedTotal: Int,
+                                        forceKillOldExecutors: Boolean,
+  newMemoryPerExecutorMB: Option[Int],
+  newCoresPerExecutor: Option[Int]): Future[Boolean] =
+    Future.successful(false)
   /**
    * Request that the cluster manager kill the specified executors.
    *

diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
@@ -198,6 +198,21 @@ private[spark] class StandaloneSchedulerBackend(
     }
   }
 
+  protected override def doRequestTotalExecutors
+  (requestedTotal: Int,
+   forceKillOldExecutors: Boolean,
+   newMemoryPerExecutorMB: Option[Int],
+   newCoresPerExecutor: Option[Int]):
+  Future[Boolean] = {
+    Option(client) match {
+      case Some(c) => c.requestTotalExecutors(
+        requestedTotal, forceKillOldExecutors,
+        newMemoryPerExecutorMB, newCoresPerExecutor)
+      case None =>
+        logWarning("Attempted to request executors before driver fully initialized.")
+        Future.successful(false)
+    }
+  }
   /**
    * Kill the given list of executors through the Master.
    * @return whether the kill request is acknowledged.

diff --git a/examples/pom.xml b/examples/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/external/avro/pom.xml b/external/avro/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/external/docker-integration-tests/pom.xml b/external/docker-integration-tests/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/external/flume-assembly/pom.xml b/external/flume-assembly/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/external/flume-sink/pom.xml b/external/flume-sink/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/external/flume/pom.xml b/external/flume/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

diff --git a/external/kafka-0-10-assembly/pom.xml b/external/kafka-0-10-assembly/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.4.1-kylin-r48</version>
+    <version>2.4.1-kylin-r51</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>