You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by va...@apache.org on 2015/12/03 19:33:19 UTC
spark git commit: [SPARK-11314][YARN] add service API and test
service for Yarn Cluster schedulers
Repository: spark
Updated Branches:
refs/heads/master 43c575cb1 -> 8fa3e474a
[SPARK-11314][YARN] add service API and test service for Yarn Cluster schedulers
This is purely the yarn/src/main and yarn/src/test bits of the YARN ATS integration: the extension model to load and run implementations of `SchedulerExtensionService` in the yarn cluster scheduler process —and to stop them afterwards.
There's duplication between the two schedulers, yarn-client and yarn-cluster, at least in terms of setting everything up, because the common superclass, `YarnSchedulerBackend` is in spark-core, and the extension services need the YARN app/attempt IDs.
If you look at how the the extension services are loaded, the case class `SchedulerExtensionServiceBinding` is used to pass in config info -currently just the spark context and the yarn IDs, of which one, the attemptID, will be null when running client-side. I'm passing in a case class to ensure that it would be possible in future to add extra arguments to the binding class, yet, as the method signature will not have changed, still be able to load existing services.
There's no functional extension service here, just one for testing. The real tests come in the bigger pull requests. At the same time, there's no restriction of this extension service purely to the ATS history publisher. Anything else that wants to listen to the spark context and publish events could use this, and I'd also consider writing one for the YARN-913 registry service, so that the URLs of the web UI would be locatable through that (low priority; would make more sense if integrated with a REST client).
There's no minicluster test. Given the test execution overhead of setting up minicluster tests, it'd probably be better to add an extension service into one of the existing tests.
Author: Steve Loughran <st...@hortonworks.com>
Closes #9182 from steveloughran/stevel/feature/SPARK-1537-service.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/8fa3e474
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/8fa3e474
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/8fa3e474
Branch: refs/heads/master
Commit: 8fa3e474a8ba180188361c0ad7e2704c3e2258d3
Parents: 43c575c
Author: Steve Loughran <st...@hortonworks.com>
Authored: Thu Dec 3 10:33:06 2015 -0800
Committer: Marcelo Vanzin <va...@cloudera.com>
Committed: Thu Dec 3 10:33:06 2015 -0800
----------------------------------------------------------------------
.../cluster/YarnSchedulerBackend.scala | 226 ---------------
.../spark/deploy/yarn/ApplicationMaster.scala | 8 +
.../cluster/SchedulerExtensionService.scala | 154 ++++++++++
.../cluster/YarnClientSchedulerBackend.scala | 22 +-
.../cluster/YarnClusterSchedulerBackend.scala | 20 +-
.../cluster/YarnSchedulerBackend.scala | 284 +++++++++++++++++++
.../ExtensionServiceIntegrationSuite.scala | 71 +++++
.../cluster/SimpleExtensionService.scala | 34 +++
.../cluster/StubApplicationAttemptId.scala | 48 ++++
.../scheduler/cluster/StubApplicationId.scala | 42 +++
10 files changed, 651 insertions(+), 258 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/core/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
deleted file mode 100644
index 80da37b..0000000
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
+++ /dev/null
@@ -1,226 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.scheduler.cluster
-
-import scala.collection.mutable.ArrayBuffer
-import scala.concurrent.{Future, ExecutionContext}
-
-import org.apache.spark.{Logging, SparkContext}
-import org.apache.spark.rpc._
-import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
-import org.apache.spark.scheduler._
-import org.apache.spark.ui.JettyUtils
-import org.apache.spark.util.{ThreadUtils, RpcUtils}
-
-import scala.util.control.NonFatal
-
-/**
- * Abstract Yarn scheduler backend that contains common logic
- * between the client and cluster Yarn scheduler backends.
- */
-private[spark] abstract class YarnSchedulerBackend(
- scheduler: TaskSchedulerImpl,
- sc: SparkContext)
- extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
-
- if (conf.getOption("spark.scheduler.minRegisteredResourcesRatio").isEmpty) {
- minRegisteredRatio = 0.8
- }
-
- protected var totalExpectedExecutors = 0
-
- private val yarnSchedulerEndpoint = new YarnSchedulerEndpoint(rpcEnv)
-
- private val yarnSchedulerEndpointRef = rpcEnv.setupEndpoint(
- YarnSchedulerBackend.ENDPOINT_NAME, yarnSchedulerEndpoint)
-
- private implicit val askTimeout = RpcUtils.askRpcTimeout(sc.conf)
-
- /**
- * Request executors from the ApplicationMaster by specifying the total number desired.
- * This includes executors already pending or running.
- */
- override def doRequestTotalExecutors(requestedTotal: Int): Boolean = {
- yarnSchedulerEndpointRef.askWithRetry[Boolean](
- RequestExecutors(requestedTotal, localityAwareTasks, hostToLocalTaskCount))
- }
-
- /**
- * Request that the ApplicationMaster kill the specified executors.
- */
- override def doKillExecutors(executorIds: Seq[String]): Boolean = {
- yarnSchedulerEndpointRef.askWithRetry[Boolean](KillExecutors(executorIds))
- }
-
- override def sufficientResourcesRegistered(): Boolean = {
- totalRegisteredExecutors.get() >= totalExpectedExecutors * minRegisteredRatio
- }
-
- /**
- * Add filters to the SparkUI.
- */
- private def addWebUIFilter(
- filterName: String,
- filterParams: Map[String, String],
- proxyBase: String): Unit = {
- if (proxyBase != null && proxyBase.nonEmpty) {
- System.setProperty("spark.ui.proxyBase", proxyBase)
- }
-
- val hasFilter =
- filterName != null && filterName.nonEmpty &&
- filterParams != null && filterParams.nonEmpty
- if (hasFilter) {
- logInfo(s"Add WebUI Filter. $filterName, $filterParams, $proxyBase")
- conf.set("spark.ui.filters", filterName)
- filterParams.foreach { case (k, v) => conf.set(s"spark.$filterName.param.$k", v) }
- scheduler.sc.ui.foreach { ui => JettyUtils.addFilters(ui.getHandlers, conf) }
- }
- }
-
- override def createDriverEndpoint(properties: Seq[(String, String)]): DriverEndpoint = {
- new YarnDriverEndpoint(rpcEnv, properties)
- }
-
- /**
- * Override the DriverEndpoint to add extra logic for the case when an executor is disconnected.
- * This endpoint communicates with the executors and queries the AM for an executor's exit
- * status when the executor is disconnected.
- */
- private class YarnDriverEndpoint(rpcEnv: RpcEnv, sparkProperties: Seq[(String, String)])
- extends DriverEndpoint(rpcEnv, sparkProperties) {
-
- /**
- * When onDisconnected is received at the driver endpoint, the superclass DriverEndpoint
- * handles it by assuming the Executor was lost for a bad reason and removes the executor
- * immediately.
- *
- * In YARN's case however it is crucial to talk to the application master and ask why the
- * executor had exited. If the executor exited for some reason unrelated to the running tasks
- * (e.g., preemption), according to the application master, then we pass that information down
- * to the TaskSetManager to inform the TaskSetManager that tasks on that lost executor should
- * not count towards a job failure.
- */
- override def onDisconnected(rpcAddress: RpcAddress): Unit = {
- addressToExecutorId.get(rpcAddress).foreach { executorId =>
- if (disableExecutor(executorId)) {
- yarnSchedulerEndpoint.handleExecutorDisconnectedFromDriver(executorId, rpcAddress)
- }
- }
- }
- }
-
- /**
- * An [[RpcEndpoint]] that communicates with the ApplicationMaster.
- */
- private class YarnSchedulerEndpoint(override val rpcEnv: RpcEnv)
- extends ThreadSafeRpcEndpoint with Logging {
- private var amEndpoint: Option[RpcEndpointRef] = None
-
- private val askAmThreadPool =
- ThreadUtils.newDaemonCachedThreadPool("yarn-scheduler-ask-am-thread-pool")
- implicit val askAmExecutor = ExecutionContext.fromExecutor(askAmThreadPool)
-
- private[YarnSchedulerBackend] def handleExecutorDisconnectedFromDriver(
- executorId: String,
- executorRpcAddress: RpcAddress): Unit = {
- amEndpoint match {
- case Some(am) =>
- val lossReasonRequest = GetExecutorLossReason(executorId)
- val future = am.ask[ExecutorLossReason](lossReasonRequest, askTimeout)
- future onSuccess {
- case reason: ExecutorLossReason => {
- driverEndpoint.askWithRetry[Boolean](RemoveExecutor(executorId, reason))
- }
- }
- future onFailure {
- case NonFatal(e) => {
- logWarning(s"Attempted to get executor loss reason" +
- s" for executor id ${executorId} at RPC address ${executorRpcAddress}," +
- s" but got no response. Marking as slave lost.", e)
- driverEndpoint.askWithRetry[Boolean](RemoveExecutor(executorId, SlaveLost()))
- }
- case t => throw t
- }
- case None =>
- logWarning("Attempted to check for an executor loss reason" +
- " before the AM has registered!")
- }
- }
-
- override def receive: PartialFunction[Any, Unit] = {
- case RegisterClusterManager(am) =>
- logInfo(s"ApplicationMaster registered as $am")
- amEndpoint = Option(am)
-
- case AddWebUIFilter(filterName, filterParams, proxyBase) =>
- addWebUIFilter(filterName, filterParams, proxyBase)
-
- case RemoveExecutor(executorId, reason) =>
- logWarning(reason.toString)
- removeExecutor(executorId, reason)
- }
-
-
- override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
- case r: RequestExecutors =>
- amEndpoint match {
- case Some(am) =>
- Future {
- context.reply(am.askWithRetry[Boolean](r))
- } onFailure {
- case NonFatal(e) =>
- logError(s"Sending $r to AM was unsuccessful", e)
- context.sendFailure(e)
- }
- case None =>
- logWarning("Attempted to request executors before the AM has registered!")
- context.reply(false)
- }
-
- case k: KillExecutors =>
- amEndpoint match {
- case Some(am) =>
- Future {
- context.reply(am.askWithRetry[Boolean](k))
- } onFailure {
- case NonFatal(e) =>
- logError(s"Sending $k to AM was unsuccessful", e)
- context.sendFailure(e)
- }
- case None =>
- logWarning("Attempted to kill executors before the AM has registered!")
- context.reply(false)
- }
- }
-
- override def onDisconnected(remoteAddress: RpcAddress): Unit = {
- if (amEndpoint.exists(_.address == remoteAddress)) {
- logWarning(s"ApplicationMaster has disassociated: $remoteAddress")
- }
- }
-
- override def onStop(): Unit = {
- askAmThreadPool.shutdownNow()
- }
- }
-}
-
-private[spark] object YarnSchedulerBackend {
- val ENDPOINT_NAME = "YarnScheduler"
-}
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
----------------------------------------------------------------------
diff --git a/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala b/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
index 50ae7ff..13ef4df 100644
--- a/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
+++ b/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
@@ -117,6 +117,10 @@ private[spark] class ApplicationMaster(
private var delegationTokenRenewerOption: Option[AMDelegationTokenRenewer] = None
+ def getAttemptId(): ApplicationAttemptId = {
+ client.getAttemptId()
+ }
+
final def run(): Int = {
try {
val appAttemptId = client.getAttemptId()
@@ -662,6 +666,10 @@ object ApplicationMaster extends Logging {
master.sparkContextStopped(sc)
}
+ private[spark] def getAttemptId(): ApplicationAttemptId = {
+ master.getAttemptId
+ }
+
}
/**
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/yarn/src/main/scala/org/apache/spark/scheduler/cluster/SchedulerExtensionService.scala
----------------------------------------------------------------------
diff --git a/yarn/src/main/scala/org/apache/spark/scheduler/cluster/SchedulerExtensionService.scala b/yarn/src/main/scala/org/apache/spark/scheduler/cluster/SchedulerExtensionService.scala
new file mode 100644
index 0000000..c064521
--- /dev/null
+++ b/yarn/src/main/scala/org/apache/spark/scheduler/cluster/SchedulerExtensionService.scala
@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster
+
+import java.util.concurrent.atomic.AtomicBoolean
+
+import org.apache.hadoop.yarn.api.records.{ApplicationAttemptId, ApplicationId}
+
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.spark.util.Utils
+
+/**
+ * An extension service that can be loaded into a Spark YARN scheduler.
+ * A Service that can be started and stopped.
+ *
+ * 1. For implementations to be loadable by `SchedulerExtensionServices`,
+ * they must provide an empty constructor.
+ * 2. The `stop()` operation MUST be idempotent, and succeed even if `start()` was
+ * never invoked.
+ */
+trait SchedulerExtensionService {
+
+ /**
+ * Start the extension service. This should be a no-op if
+ * called more than once.
+ * @param binding binding to the spark application and YARN
+ */
+ def start(binding: SchedulerExtensionServiceBinding): Unit
+
+ /**
+ * Stop the service
+ * The `stop()` operation MUST be idempotent, and succeed even if `start()` was
+ * never invoked.
+ */
+ def stop(): Unit
+}
+
+/**
+ * Binding information for a [[SchedulerExtensionService]].
+ *
+ * The attempt ID will be set if the service is started within a YARN application master;
+ * there is then a different attempt ID for every time that AM is restarted.
+ * When the service binding is instantiated in client mode, there's no attempt ID, as it lacks
+ * this information.
+ * @param sparkContext current spark context
+ * @param applicationId YARN application ID
+ * @param attemptId YARN attemptID. This will always be unset in client mode, and always set in
+ * cluster mode.
+ */
+case class SchedulerExtensionServiceBinding(
+ sparkContext: SparkContext,
+ applicationId: ApplicationId,
+ attemptId: Option[ApplicationAttemptId] = None)
+
+/**
+ * Container for [[SchedulerExtensionService]] instances.
+ *
+ * Loads Extension Services from the configuration property
+ * `"spark.yarn.services"`, instantiates and starts them.
+ * When stopped, it stops all child entries.
+ *
+ * The order in which child extension services are started and stopped
+ * is undefined.
+ */
+private[spark] class SchedulerExtensionServices extends SchedulerExtensionService
+ with Logging {
+ private var serviceOption: Option[String] = None
+ private var services: List[SchedulerExtensionService] = Nil
+ private val started = new AtomicBoolean(false)
+ private var binding: SchedulerExtensionServiceBinding = _
+
+ /**
+ * Binding operation will load the named services and call bind on them too; the
+ * entire set of services are then ready for `init()` and `start()` calls.
+ *
+ * @param binding binding to the spark application and YARN
+ */
+ def start(binding: SchedulerExtensionServiceBinding): Unit = {
+ if (started.getAndSet(true)) {
+ logWarning("Ignoring re-entrant start operation")
+ return
+ }
+ require(binding.sparkContext != null, "Null context parameter")
+ require(binding.applicationId != null, "Null appId parameter")
+ this.binding = binding
+ val sparkContext = binding.sparkContext
+ val appId = binding.applicationId
+ val attemptId = binding.attemptId
+ logInfo(s"Starting Yarn extension services with app $appId and attemptId $attemptId")
+
+ serviceOption = sparkContext.getConf.getOption(SchedulerExtensionServices.SPARK_YARN_SERVICES)
+ services = serviceOption
+ .map { s =>
+ s.split(",").map(_.trim()).filter(!_.isEmpty)
+ .map { sClass =>
+ val instance = Utils.classForName(sClass)
+ .newInstance()
+ .asInstanceOf[SchedulerExtensionService]
+ // bind this service
+ instance.start(binding)
+ logInfo(s"Service $sClass started")
+ instance
+ }.toList
+ }.getOrElse(Nil)
+ }
+
+ /**
+ * Get the list of services.
+ *
+ * @return a list of services; Nil until the service is started
+ */
+ def getServices: List[SchedulerExtensionService] = services
+
+ /**
+ * Stop the services; idempotent.
+ *
+ */
+ override def stop(): Unit = {
+ if (started.getAndSet(false)) {
+ logInfo(s"Stopping $this")
+ services.foreach { s =>
+ Utils.tryLogNonFatalError(s.stop())
+ }
+ }
+ }
+
+ override def toString(): String = s"""SchedulerExtensionServices
+ |(serviceOption=$serviceOption,
+ | services=$services,
+ | started=$started)""".stripMargin
+}
+
+private[spark] object SchedulerExtensionServices {
+
+ /**
+ * A list of comma separated services to instantiate in the scheduler
+ */
+ val SPARK_YARN_SERVICES = "spark.yarn.services"
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala
----------------------------------------------------------------------
diff --git a/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala b/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala
index 20771f6..0e27a26 100644
--- a/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala
+++ b/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala
@@ -19,7 +19,7 @@ package org.apache.spark.scheduler.cluster
import scala.collection.mutable.ArrayBuffer
-import org.apache.hadoop.yarn.api.records.{ApplicationId, YarnApplicationState}
+import org.apache.hadoop.yarn.api.records.YarnApplicationState
import org.apache.spark.{SparkException, Logging, SparkContext}
import org.apache.spark.deploy.yarn.{Client, ClientArguments, YarnSparkHadoopUtil}
@@ -33,7 +33,6 @@ private[spark] class YarnClientSchedulerBackend(
with Logging {
private var client: Client = null
- private var appId: ApplicationId = null
private var monitorThread: MonitorThread = null
/**
@@ -54,13 +53,12 @@ private[spark] class YarnClientSchedulerBackend(
val args = new ClientArguments(argsArrayBuf.toArray, conf)
totalExpectedExecutors = args.numExecutors
client = new Client(args, conf)
- appId = client.submitApplication()
+ bindToYarn(client.submitApplication(), None)
// SPARK-8687: Ensure all necessary properties have already been set before
// we initialize our driver scheduler backend, which serves these properties
// to the executors
super.start()
-
waitForApplication()
// SPARK-8851: In yarn-client mode, the AM still does the credentials refresh. The driver
@@ -116,8 +114,8 @@ private[spark] class YarnClientSchedulerBackend(
* This assumes both `client` and `appId` have already been set.
*/
private def waitForApplication(): Unit = {
- assert(client != null && appId != null, "Application has not been submitted yet!")
- val (state, _) = client.monitorApplication(appId, returnOnRunning = true) // blocking
+ assert(client != null && appId.isDefined, "Application has not been submitted yet!")
+ val (state, _) = client.monitorApplication(appId.get, returnOnRunning = true) // blocking
if (state == YarnApplicationState.FINISHED ||
state == YarnApplicationState.FAILED ||
state == YarnApplicationState.KILLED) {
@@ -125,7 +123,7 @@ private[spark] class YarnClientSchedulerBackend(
"It might have been killed or unable to launch application master.")
}
if (state == YarnApplicationState.RUNNING) {
- logInfo(s"Application $appId has started running.")
+ logInfo(s"Application ${appId.get} has started running.")
}
}
@@ -141,7 +139,7 @@ private[spark] class YarnClientSchedulerBackend(
override def run() {
try {
- val (state, _) = client.monitorApplication(appId, logApplicationReport = false)
+ val (state, _) = client.monitorApplication(appId.get, logApplicationReport = false)
logError(s"Yarn application has already exited with state $state!")
allowInterrupt = false
sc.stop()
@@ -163,7 +161,7 @@ private[spark] class YarnClientSchedulerBackend(
* This assumes both `client` and `appId` have already been set.
*/
private def asyncMonitorApplication(): MonitorThread = {
- assert(client != null && appId != null, "Application has not been submitted yet!")
+ assert(client != null && appId.isDefined, "Application has not been submitted yet!")
val t = new MonitorThread
t.setName("Yarn application state monitor")
t.setDaemon(true)
@@ -193,10 +191,4 @@ private[spark] class YarnClientSchedulerBackend(
logInfo("Stopped")
}
- override def applicationId(): String = {
- Option(appId).map(_.toString).getOrElse {
- logWarning("Application ID is not initialized yet.")
- super.applicationId
- }
- }
}
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClusterSchedulerBackend.scala
----------------------------------------------------------------------
diff --git a/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClusterSchedulerBackend.scala b/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClusterSchedulerBackend.scala
index 50b699f..ced597b 100644
--- a/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClusterSchedulerBackend.scala
+++ b/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnClusterSchedulerBackend.scala
@@ -21,7 +21,7 @@ import org.apache.hadoop.yarn.api.ApplicationConstants.Environment
import org.apache.hadoop.yarn.conf.YarnConfiguration
import org.apache.spark.SparkContext
-import org.apache.spark.deploy.yarn.YarnSparkHadoopUtil
+import org.apache.spark.deploy.yarn.{ApplicationMaster, YarnSparkHadoopUtil}
import org.apache.spark.scheduler.TaskSchedulerImpl
import org.apache.spark.util.Utils
@@ -31,26 +31,12 @@ private[spark] class YarnClusterSchedulerBackend(
extends YarnSchedulerBackend(scheduler, sc) {
override def start() {
+ val attemptId = ApplicationMaster.getAttemptId
+ bindToYarn(attemptId.getApplicationId(), Some(attemptId))
super.start()
totalExpectedExecutors = YarnSparkHadoopUtil.getInitialTargetExecutorNumber(sc.conf)
}
- override def applicationId(): String =
- // In YARN Cluster mode, the application ID is expected to be set, so log an error if it's
- // not found.
- sc.getConf.getOption("spark.yarn.app.id").getOrElse {
- logError("Application ID is not set.")
- super.applicationId
- }
-
- override def applicationAttemptId(): Option[String] =
- // In YARN Cluster mode, the attempt ID is expected to be set, so log an error if it's
- // not found.
- sc.getConf.getOption("spark.yarn.app.attemptId").orElse {
- logError("Application attempt ID is not set.")
- super.applicationAttemptId
- }
-
override def getDriverLogUrls: Option[Map[String, String]] = {
var driverLogs: Option[Map[String, String]] = None
try {
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
----------------------------------------------------------------------
diff --git a/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala b/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
new file mode 100644
index 0000000..e3dd877
--- /dev/null
+++ b/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.util.control.NonFatal
+
+import org.apache.hadoop.yarn.api.records.{ApplicationAttemptId, ApplicationId}
+
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.spark.rpc._
+import org.apache.spark.scheduler._
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
+import org.apache.spark.ui.JettyUtils
+import org.apache.spark.util.{RpcUtils, ThreadUtils}
+
+/**
+ * Abstract Yarn scheduler backend that contains common logic
+ * between the client and cluster Yarn scheduler backends.
+ */
+private[spark] abstract class YarnSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ sc: SparkContext)
+ extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
+
+ if (conf.getOption("spark.scheduler.minRegisteredResourcesRatio").isEmpty) {
+ minRegisteredRatio = 0.8
+ }
+
+ protected var totalExpectedExecutors = 0
+
+ private val yarnSchedulerEndpoint = new YarnSchedulerEndpoint(rpcEnv)
+
+ private val yarnSchedulerEndpointRef = rpcEnv.setupEndpoint(
+ YarnSchedulerBackend.ENDPOINT_NAME, yarnSchedulerEndpoint)
+
+ private implicit val askTimeout = RpcUtils.askRpcTimeout(sc.conf)
+
+ /** Application ID. */
+ protected var appId: Option[ApplicationId] = None
+
+ /** Attempt ID. This is unset for client-mode schedulers */
+ private var attemptId: Option[ApplicationAttemptId] = None
+
+ /** Scheduler extension services. */
+ private val services: SchedulerExtensionServices = new SchedulerExtensionServices()
+
+ /**
+ * Bind to YARN. This *must* be done before calling [[start()]].
+ *
+ * @param appId YARN application ID
+ * @param attemptId Optional YARN attempt ID
+ */
+ protected def bindToYarn(appId: ApplicationId, attemptId: Option[ApplicationAttemptId]): Unit = {
+ this.appId = Some(appId)
+ this.attemptId = attemptId
+ }
+
+ override def start() {
+ require(appId.isDefined, "application ID unset")
+ val binding = SchedulerExtensionServiceBinding(sc, appId.get, attemptId)
+ services.start(binding)
+ super.start()
+ }
+
+ override def stop(): Unit = {
+ try {
+ super.stop()
+ } finally {
+ services.stop()
+ }
+ }
+
+ /**
+ * Get the attempt ID for this run, if the cluster manager supports multiple
+ * attempts. Applications run in client mode will not have attempt IDs.
+ *
+ * @return The application attempt id, if available.
+ */
+ override def applicationAttemptId(): Option[String] = {
+ attemptId.map(_.toString)
+ }
+
+ /**
+ * Get an application ID associated with the job.
+ * This returns the string value of [[appId]] if set, otherwise
+ * the locally-generated ID from the superclass.
+ * @return The application ID
+ */
+ override def applicationId(): String = {
+ appId.map(_.toString).getOrElse {
+ logWarning("Application ID is not initialized yet.")
+ super.applicationId
+ }
+ }
+
+ /**
+ * Request executors from the ApplicationMaster by specifying the total number desired.
+ * This includes executors already pending or running.
+ */
+ override def doRequestTotalExecutors(requestedTotal: Int): Boolean = {
+ yarnSchedulerEndpointRef.askWithRetry[Boolean](
+ RequestExecutors(requestedTotal, localityAwareTasks, hostToLocalTaskCount))
+ }
+
+ /**
+ * Request that the ApplicationMaster kill the specified executors.
+ */
+ override def doKillExecutors(executorIds: Seq[String]): Boolean = {
+ yarnSchedulerEndpointRef.askWithRetry[Boolean](KillExecutors(executorIds))
+ }
+
+ override def sufficientResourcesRegistered(): Boolean = {
+ totalRegisteredExecutors.get() >= totalExpectedExecutors * minRegisteredRatio
+ }
+
+ /**
+ * Add filters to the SparkUI.
+ */
+ private def addWebUIFilter(
+ filterName: String,
+ filterParams: Map[String, String],
+ proxyBase: String): Unit = {
+ if (proxyBase != null && proxyBase.nonEmpty) {
+ System.setProperty("spark.ui.proxyBase", proxyBase)
+ }
+
+ val hasFilter =
+ filterName != null && filterName.nonEmpty &&
+ filterParams != null && filterParams.nonEmpty
+ if (hasFilter) {
+ logInfo(s"Add WebUI Filter. $filterName, $filterParams, $proxyBase")
+ conf.set("spark.ui.filters", filterName)
+ filterParams.foreach { case (k, v) => conf.set(s"spark.$filterName.param.$k", v) }
+ scheduler.sc.ui.foreach { ui => JettyUtils.addFilters(ui.getHandlers, conf) }
+ }
+ }
+
+ override def createDriverEndpoint(properties: Seq[(String, String)]): DriverEndpoint = {
+ new YarnDriverEndpoint(rpcEnv, properties)
+ }
+
+ /**
+ * Override the DriverEndpoint to add extra logic for the case when an executor is disconnected.
+ * This endpoint communicates with the executors and queries the AM for an executor's exit
+ * status when the executor is disconnected.
+ */
+ private class YarnDriverEndpoint(rpcEnv: RpcEnv, sparkProperties: Seq[(String, String)])
+ extends DriverEndpoint(rpcEnv, sparkProperties) {
+
+ /**
+ * When onDisconnected is received at the driver endpoint, the superclass DriverEndpoint
+ * handles it by assuming the Executor was lost for a bad reason and removes the executor
+ * immediately.
+ *
+ * In YARN's case however it is crucial to talk to the application master and ask why the
+ * executor had exited. If the executor exited for some reason unrelated to the running tasks
+ * (e.g., preemption), according to the application master, then we pass that information down
+ * to the TaskSetManager to inform the TaskSetManager that tasks on that lost executor should
+ * not count towards a job failure.
+ */
+ override def onDisconnected(rpcAddress: RpcAddress): Unit = {
+ addressToExecutorId.get(rpcAddress).foreach { executorId =>
+ if (disableExecutor(executorId)) {
+ yarnSchedulerEndpoint.handleExecutorDisconnectedFromDriver(executorId, rpcAddress)
+ }
+ }
+ }
+ }
+
+ /**
+ * An [[RpcEndpoint]] that communicates with the ApplicationMaster.
+ */
+ private class YarnSchedulerEndpoint(override val rpcEnv: RpcEnv)
+ extends ThreadSafeRpcEndpoint with Logging {
+ private var amEndpoint: Option[RpcEndpointRef] = None
+
+ private val askAmThreadPool =
+ ThreadUtils.newDaemonCachedThreadPool("yarn-scheduler-ask-am-thread-pool")
+ implicit val askAmExecutor = ExecutionContext.fromExecutor(askAmThreadPool)
+
+ private[YarnSchedulerBackend] def handleExecutorDisconnectedFromDriver(
+ executorId: String,
+ executorRpcAddress: RpcAddress): Unit = {
+ amEndpoint match {
+ case Some(am) =>
+ val lossReasonRequest = GetExecutorLossReason(executorId)
+ val future = am.ask[ExecutorLossReason](lossReasonRequest, askTimeout)
+ future onSuccess {
+ case reason: ExecutorLossReason => {
+ driverEndpoint.askWithRetry[Boolean](RemoveExecutor(executorId, reason))
+ }
+ }
+ future onFailure {
+ case NonFatal(e) => {
+ logWarning(s"Attempted to get executor loss reason" +
+ s" for executor id ${executorId} at RPC address ${executorRpcAddress}," +
+ s" but got no response. Marking as slave lost.", e)
+ driverEndpoint.askWithRetry[Boolean](RemoveExecutor(executorId, SlaveLost()))
+ }
+ case t => throw t
+ }
+ case None =>
+ logWarning("Attempted to check for an executor loss reason" +
+ " before the AM has registered!")
+ }
+ }
+
+ override def receive: PartialFunction[Any, Unit] = {
+ case RegisterClusterManager(am) =>
+ logInfo(s"ApplicationMaster registered as $am")
+ amEndpoint = Option(am)
+
+ case AddWebUIFilter(filterName, filterParams, proxyBase) =>
+ addWebUIFilter(filterName, filterParams, proxyBase)
+
+ case RemoveExecutor(executorId, reason) =>
+ logWarning(reason.toString)
+ removeExecutor(executorId, reason)
+ }
+
+
+ override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
+ case r: RequestExecutors =>
+ amEndpoint match {
+ case Some(am) =>
+ Future {
+ context.reply(am.askWithRetry[Boolean](r))
+ } onFailure {
+ case NonFatal(e) =>
+ logError(s"Sending $r to AM was unsuccessful", e)
+ context.sendFailure(e)
+ }
+ case None =>
+ logWarning("Attempted to request executors before the AM has registered!")
+ context.reply(false)
+ }
+
+ case k: KillExecutors =>
+ amEndpoint match {
+ case Some(am) =>
+ Future {
+ context.reply(am.askWithRetry[Boolean](k))
+ } onFailure {
+ case NonFatal(e) =>
+ logError(s"Sending $k to AM was unsuccessful", e)
+ context.sendFailure(e)
+ }
+ case None =>
+ logWarning("Attempted to kill executors before the AM has registered!")
+ context.reply(false)
+ }
+ }
+
+ override def onDisconnected(remoteAddress: RpcAddress): Unit = {
+ if (amEndpoint.exists(_.address == remoteAddress)) {
+ logWarning(s"ApplicationMaster has disassociated: $remoteAddress")
+ }
+ }
+
+ override def onStop(): Unit = {
+ askAmThreadPool.shutdownNow()
+ }
+ }
+}
+
+private[spark] object YarnSchedulerBackend {
+ val ENDPOINT_NAME = "YarnScheduler"
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/yarn/src/test/scala/org/apache/spark/scheduler/cluster/ExtensionServiceIntegrationSuite.scala
----------------------------------------------------------------------
diff --git a/yarn/src/test/scala/org/apache/spark/scheduler/cluster/ExtensionServiceIntegrationSuite.scala b/yarn/src/test/scala/org/apache/spark/scheduler/cluster/ExtensionServiceIntegrationSuite.scala
new file mode 100644
index 0000000..b4d1b0a
--- /dev/null
+++ b/yarn/src/test/scala/org/apache/spark/scheduler/cluster/ExtensionServiceIntegrationSuite.scala
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.{LocalSparkContext, Logging, SparkConf, SparkContext, SparkFunSuite}
+
+/**
+ * Test the integration with [[SchedulerExtensionServices]]
+ */
+class ExtensionServiceIntegrationSuite extends SparkFunSuite
+ with LocalSparkContext with BeforeAndAfter
+ with Logging {
+
+ val applicationId = new StubApplicationId(0, 1111L)
+ val attemptId = new StubApplicationAttemptId(applicationId, 1)
+
+ /*
+ * Setup phase creates the spark context
+ */
+ before {
+ val sparkConf = new SparkConf()
+ sparkConf.set(SchedulerExtensionServices.SPARK_YARN_SERVICES,
+ classOf[SimpleExtensionService].getName())
+ sparkConf.setMaster("local").setAppName("ExtensionServiceIntegrationSuite")
+ sc = new SparkContext(sparkConf)
+ }
+
+ test("Instantiate") {
+ val services = new SchedulerExtensionServices()
+ assertResult(Nil, "non-nil service list") {
+ services.getServices
+ }
+ services.start(SchedulerExtensionServiceBinding(sc, applicationId))
+ services.stop()
+ }
+
+ test("Contains SimpleExtensionService Service") {
+ val services = new SchedulerExtensionServices()
+ try {
+ services.start(SchedulerExtensionServiceBinding(sc, applicationId))
+ val serviceList = services.getServices
+ assert(serviceList.nonEmpty, "empty service list")
+ val (service :: Nil) = serviceList
+ val simpleService = service.asInstanceOf[SimpleExtensionService]
+ assert(simpleService.started.get, "service not started")
+ services.stop()
+ assert(!simpleService.started.get, "service not stopped")
+ } finally {
+ services.stop()
+ }
+ }
+}
+
+
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/yarn/src/test/scala/org/apache/spark/scheduler/cluster/SimpleExtensionService.scala
----------------------------------------------------------------------
diff --git a/yarn/src/test/scala/org/apache/spark/scheduler/cluster/SimpleExtensionService.scala b/yarn/src/test/scala/org/apache/spark/scheduler/cluster/SimpleExtensionService.scala
new file mode 100644
index 0000000..9b8c98c
--- /dev/null
+++ b/yarn/src/test/scala/org/apache/spark/scheduler/cluster/SimpleExtensionService.scala
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster
+
+import java.util.concurrent.atomic.AtomicBoolean
+
+private[spark] class SimpleExtensionService extends SchedulerExtensionService {
+
+ /** started flag; set in the `start()` call, stopped in `stop()`. */
+ val started = new AtomicBoolean(false)
+
+ override def start(binding: SchedulerExtensionServiceBinding): Unit = {
+ started.set(true)
+ }
+
+ override def stop(): Unit = {
+ started.set(false)
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/yarn/src/test/scala/org/apache/spark/scheduler/cluster/StubApplicationAttemptId.scala
----------------------------------------------------------------------
diff --git a/yarn/src/test/scala/org/apache/spark/scheduler/cluster/StubApplicationAttemptId.scala b/yarn/src/test/scala/org/apache/spark/scheduler/cluster/StubApplicationAttemptId.scala
new file mode 100644
index 0000000..4b57b95
--- /dev/null
+++ b/yarn/src/test/scala/org/apache/spark/scheduler/cluster/StubApplicationAttemptId.scala
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster
+
+import org.apache.hadoop.yarn.api.records.{ApplicationAttemptId, ApplicationId}
+
+/**
+ * A stub application ID; can be set in constructor and/or updated later.
+ * @param applicationId application ID
+ * @param attempt an attempt counter
+ */
+class StubApplicationAttemptId(var applicationId: ApplicationId, var attempt: Int)
+ extends ApplicationAttemptId {
+
+ override def setApplicationId(appID: ApplicationId): Unit = {
+ applicationId = appID
+ }
+
+ override def getAttemptId: Int = {
+ attempt
+ }
+
+ override def setAttemptId(attemptId: Int): Unit = {
+ attempt = attemptId
+ }
+
+ override def getApplicationId: ApplicationId = {
+ applicationId
+ }
+
+ override def build(): Unit = {
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/8fa3e474/yarn/src/test/scala/org/apache/spark/scheduler/cluster/StubApplicationId.scala
----------------------------------------------------------------------
diff --git a/yarn/src/test/scala/org/apache/spark/scheduler/cluster/StubApplicationId.scala b/yarn/src/test/scala/org/apache/spark/scheduler/cluster/StubApplicationId.scala
new file mode 100644
index 0000000..bffa0e0
--- /dev/null
+++ b/yarn/src/test/scala/org/apache/spark/scheduler/cluster/StubApplicationId.scala
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler.cluster
+
+import org.apache.hadoop.yarn.api.records.ApplicationId
+
+/**
+ * Simple Testing Application Id; ID and cluster timestamp are set in constructor
+ * and cannot be updated.
+ * @param id app id
+ * @param clusterTimestamp timestamp
+ */
+private[spark] class StubApplicationId(id: Int, clusterTimestamp: Long) extends ApplicationId {
+ override def getId: Int = {
+ id
+ }
+
+ override def getClusterTimestamp: Long = {
+ clusterTimestamp
+ }
+
+ override def setId(id: Int): Unit = {}
+
+ override def setClusterTimestamp(clusterTimestamp: Long): Unit = {}
+
+ override def build(): Unit = {}
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org