Skip to content

[SPARK-22404][YARN] Provide an option to use unmanaged AM in yarn-client mode #19616

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
devaraj-kavali wants to merge 16 commits into from
Closed

[SPARK-22404][YARN] Provide an option to use unmanaged AM in yarn-client mode #19616

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
e51f99e
[SPARK-22404][YARN] Provide an option to use unmanaged AM in yarn-client
Oct 31, 2017
764c302
Merge branch 'master' into SPARK-22404
Dec 13, 2017
640013b
Adding Client changes missed while merging master
Dec 13, 2017
cba0c6d
Resolving ApplicationMaster conflicts with the merge
Dec 14, 2017
19b6c3a
Fixed the review comments
Feb 15, 2018
ce94235
Merge branch 'master' into SPARK-22404
Feb 15, 2018
0921f7a
Merge branch 'master' into SPARK-22404
Jul 16, 2018
837d25f
Merge branch 'master' into SPARK-22404
Dec 12, 2018
65aeba9
Fixing the review comments
Dec 14, 2018
93b016f
Addressing review comments
Dec 20, 2018
dc31940
Review comments addressing
Jan 10, 2019
23ad9de
Merge branch 'master' into SPARK-22404
Jan 10, 2019
1c02b7d
Merge branch 'master' into SPARK-22404
Jan 10, 2019
2429e19
Changes to fix issues with master merge
Jan 15, 2019
6854fc4
Updated with review comments fix and added a test
Jan 24, 2019
3b377af
Updated to use YARN_UNMANAGED_AM.key
Jan 25, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
142 changes: 100 additions & 42 deletions resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import org.apache.hadoop.yarn.api._
import org.apache.hadoop.yarn.api.records._
import org.apache.hadoop.yarn.conf.YarnConfiguration
import org.apache.hadoop.yarn.exceptions.ApplicationAttemptNotFoundException
import org.apache.hadoop.yarn.server.webproxy.ProxyUriUtils
import org.apache.hadoop.yarn.util.{ConverterUtils, Records}

import org.apache.spark._
Expand All @@ -53,35 +54,20 @@ import org.apache.spark.util._
/**
* Common application master functionality for Spark on Yarn.
*/
private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends Logging {
private[spark] class ApplicationMaster(
args: ApplicationMasterArguments,
sparkConf: SparkConf,
yarnConf: YarnConfiguration) extends Logging {

// TODO: Currently, task to container is computed once (TaskSetManager) - which need not be
// optimal as more containers are available. Might need to handle this better.

private val isClusterMode = args.userClass != null

private val sparkConf = new SparkConf()
if (args.propertiesFile != null) {
Utils.getPropertiesFromFile(args.propertiesFile).foreach { case (k, v) =>
sparkConf.set(k, v)
}
}

private val securityMgr = new SecurityManager(sparkConf)

private var metricsSystem: Option[MetricsSystem] = None

// Set system properties for each config entry. This covers two use cases:
// - The default configuration stored by the SparkHadoopUtil class
// - The user application creating a new SparkConf in cluster mode
//
// Both cases create a new SparkConf object which reads these configs from system properties.
sparkConf.getAll.foreach { case (k, v) =>
sys.props(k) = v
}

private val yarnConf = new YarnConfiguration(SparkHadoopUtil.newConfiguration(sparkConf))

private val userClassLoader = {
val classpath = Client.getUserClasspath(sparkConf)
val urls = classpath.map { entry =>
Expand Down Expand Up @@ -169,8 +155,6 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
// Next wait interval before allocator poll.
private var nextAllocationInterval = initialAllocationInterval

private var rpcEnv: RpcEnv = null

// In cluster mode, used to tell the AM when the user's SparkContext has been initialized.
private val sparkContextPromise = Promise[SparkContext]()

Expand Down Expand Up @@ -294,7 +278,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
// we only want to unregister if we don't want the RM to retry
if (finalStatus == FinalApplicationStatus.SUCCEEDED || isLastAttempt) {
unregister(finalStatus, finalMsg)
cleanupStagingDir()
cleanupStagingDir(new Path(System.getenv("SPARK_YARN_STAGING_DIR")))
}
}
}
Expand Down Expand Up @@ -324,6 +308,59 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
}
}

def runUnmanaged(clientRpcEnv: RpcEnv,

@vanzin vanzin Dec 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Multi-line args start on the next line.

appAttemptId: ApplicationAttemptId,
stagingDir: Path): Unit = {
try {
new CallerContext(
"APPMASTER", sparkConf.get(APP_CALLER_CONTEXT),
Option(appAttemptId.getApplicationId.toString), None).setCurrentContext()

// This shutdown hook should run *after* the SparkContext is shut down.

@vanzin vanzin Dec 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is client mode, so you can't rely on shutdown hooks. You need to explicitly stop this service when the SparkContext is shutdown.

Imagine someone just embeds sc = new SparkContext(); ...; sc.stop() in their app code, but the app itself runs for way longer than the Spark app.

val priority = ShutdownHookManager.SPARK_CONTEXT_SHUTDOWN_PRIORITY - 1
ShutdownHookManager.addShutdownHook(priority) { () =>
if (!finished) {
finish(finalStatus, ApplicationMaster.EXIT_EARLY,
"Shutdown hook called before final status was reported.")
}

if (!unregistered) {
unregister(finalStatus, finalMsg)
cleanupStagingDir(stagingDir)
}
}

val driverRef = clientRpcEnv.setupEndpointRef(
RpcAddress(sparkConf.get("spark.driver.host"),
sparkConf.get("spark.driver.port").toInt),
YarnSchedulerBackend.ENDPOINT_NAME)
// The client-mode AM doesn't listen for incoming connections, so report an invalid port.
registerAM(Utils.localHostName, -1, sparkConf,
sparkConf.getOption("spark.driver.appUIAddress"), appAttemptId)
addAmIpFilter(Some(driverRef), ProxyUriUtils.getPath(appAttemptId.getApplicationId))
createAllocator(driverRef, sparkConf, clientRpcEnv, appAttemptId)
reporterThread.join()

} catch {
case e: Exception =>
// catch everything else if not specifically handled
logError("Uncaught exception: ", e)
finish(FinalApplicationStatus.FAILED,
ApplicationMaster.EXIT_UNCAUGHT_EXCEPTION,
"Uncaught exception: " + StringUtils.stringifyException(e))
} finally {
try {
metricsSystem.foreach { ms =>
ms.report()
ms.stop()
}
} catch {
case e: Exception =>
logWarning("Exception during stopping of the metric system: ", e)
}
}
}

/**
* Set the default final application status for client mode to UNDEFINED to handle
* if YARN HA restarts the application so that it properly retries. Set the final
Expand Down Expand Up @@ -404,18 +441,23 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
host: String,
port: Int,
_sparkConf: SparkConf,
uiAddress: Option[String]): Unit = {
val appId = client.getAttemptId().getApplicationId().toString()
val attemptId = client.getAttemptId().getAttemptId().toString()
uiAddress: Option[String],
appAttempt: ApplicationAttemptId): Unit = {
val appId = appAttempt.getApplicationId().toString()
val attemptId = appAttempt.getAttemptId().toString()
val historyAddress = ApplicationMaster
.getHistoryServerAddress(_sparkConf, yarnConf, appId, attemptId)

client.register(host, port, yarnConf, _sparkConf, uiAddress, historyAddress)
registered = true
}

private def createAllocator(driverRef: RpcEndpointRef, _sparkConf: SparkConf): Unit = {
val appId = client.getAttemptId().getApplicationId().toString()
private def createAllocator(
driverRef: RpcEndpointRef,
_sparkConf: SparkConf,
rpcEnv: RpcEnv,
appAttemptId: ApplicationAttemptId): Unit = {
val appId = appAttemptId.getApplicationId().toString()
val driverUrl = RpcEndpointAddress(driverRef.address.host, driverRef.address.port,
CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString

Expand All @@ -436,7 +478,8 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
driverUrl,
driverRef,
securityMgr,
localResources)
localResources,
appAttemptId)

tokenManager.foreach(_.setDriverRef(driverRef))

Expand All @@ -455,7 +498,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
}

private def runDriver(): Unit = {
addAmIpFilter(None)
addAmIpFilter(None, System.getenv(ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV))
userClassThread = startUserApplication()

// This a bit hacky, but we need to wait until the spark.driver.port property has
Expand All @@ -466,17 +509,17 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
val sc = ThreadUtils.awaitResult(sparkContextPromise.future,
Duration(totalWaitTime, TimeUnit.MILLISECONDS))
if (sc != null) {
rpcEnv = sc.env.rpcEnv
val rpcEnv = sc.env.rpcEnv

val userConf = sc.getConf
val host = userConf.get("spark.driver.host")
val port = userConf.get("spark.driver.port").toInt
registerAM(host, port, userConf, sc.ui.map(_.webUrl))
registerAM(host, port, userConf, sc.ui.map(_.webUrl), client.getAttemptId())

val driverRef = rpcEnv.setupEndpointRef(
RpcAddress(host, port),
YarnSchedulerBackend.ENDPOINT_NAME)
createAllocator(driverRef, userConf)
createAllocator(driverRef, userConf, rpcEnv, client.getAttemptId())
} else {
// Sanity check; should never happen in normal operation, since sc should only be null
// if the user app did not create a SparkContext.
Expand All @@ -500,20 +543,22 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
private def runExecutorLauncher(): Unit = {
val hostname = Utils.localHostName
val amCores = sparkConf.get(AM_CORES)
rpcEnv = RpcEnv.create("sparkYarnAM", hostname, hostname, -1, sparkConf, securityMgr,
val rpcEnv = RpcEnv.create("sparkYarnAM", hostname, hostname, -1, sparkConf, securityMgr,
amCores, true)

// The client-mode AM doesn't listen for incoming connections, so report an invalid port.
registerAM(hostname, -1, sparkConf, sparkConf.getOption("spark.driver.appUIAddress"))
registerAM(hostname, -1, sparkConf, sparkConf.getOption("spark.driver.appUIAddress"),
client.getAttemptId())

// The driver should be up and listening, so unlike cluster mode, just try to connect to it
// with no waiting or retrying.
val (driverHost, driverPort) = Utils.parseHostPort(args.userArgs(0))
val driverRef = rpcEnv.setupEndpointRef(
RpcAddress(driverHost, driverPort),
YarnSchedulerBackend.ENDPOINT_NAME)
addAmIpFilter(Some(driverRef))
createAllocator(driverRef, sparkConf)
addAmIpFilter(Some(driverRef),
System.getenv(ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV))
createAllocator(driverRef, sparkConf, rpcEnv, client.getAttemptId())

// In client mode the actor will stop the reporter thread.
reporterThread.join()
Expand Down Expand Up @@ -612,12 +657,10 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
/**
* Clean up the staging directory.
*/
private def cleanupStagingDir(): Unit = {
var stagingDirPath: Path = null
private def cleanupStagingDir(stagingDirPath: Path): Unit = {
try {
val preserveFiles = sparkConf.get(PRESERVE_STAGING_FILES)
if (!preserveFiles) {
stagingDirPath = new Path(System.getenv("SPARK_YARN_STAGING_DIR"))
logInfo("Deleting staging directory " + stagingDirPath)
val fs = stagingDirPath.getFileSystem(yarnConf)
fs.delete(stagingDirPath, true)
Expand All @@ -629,8 +672,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
}

/** Add the Yarn IP filter that is required for properly securing the UI. */
private def addAmIpFilter(driver: Option[RpcEndpointRef]) = {
val proxyBase = System.getenv(ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV)
private def addAmIpFilter(driver: Option[RpcEndpointRef], proxyBase: String) = {
val amFilter = "org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter"
val params = client.getAmIpFilterParams(yarnConf, proxyBase)
driver match {
Expand Down Expand Up @@ -791,7 +833,23 @@ object ApplicationMaster extends Logging {
def main(args: Array[String]): Unit = {
SignalUtils.registerLogger(log)
val amArgs = new ApplicationMasterArguments(args)
master = new ApplicationMaster(amArgs)
val sparkConf = new SparkConf()
if (amArgs.propertiesFile != null) {
Utils.getPropertiesFromFile(amArgs.propertiesFile).foreach { case (k, v) =>
sparkConf.set(k, v)
}
}
// Set system properties for each config entry. This covers two use cases:
// - The default configuration stored by the SparkHadoopUtil class
// - The user application creating a new SparkConf in cluster mode
//
// Both cases create a new SparkConf object which reads these configs from system properties.
sparkConf.getAll.foreach { case (k, v) =>
sys.props(k) = v
}

val yarnConf = new YarnConfiguration(SparkHadoopUtil.newConfiguration(sparkConf))
master = new ApplicationMaster(amArgs, sparkConf, yarnConf)
System.exit(master.run())
}

Expand Down
47 changes: 41 additions & 6 deletions resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ import com.google.common.io.Files
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._
import org.apache.hadoop.fs.permission.FsPermission
import org.apache.hadoop.io.DataOutputBuffer
import org.apache.hadoop.io.{DataOutputBuffer, Text}
import org.apache.hadoop.mapreduce.MRJobConfig
import org.apache.hadoop.security.{Credentials, UserGroupInformation}
import org.apache.hadoop.util.StringUtils
Expand All @@ -45,6 +45,7 @@ import org.apache.hadoop.yarn.api.records._
import org.apache.hadoop.yarn.client.api.{YarnClient, YarnClientApplication}
import org.apache.hadoop.yarn.conf.YarnConfiguration
import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier
import org.apache.hadoop.yarn.util.Records

import org.apache.spark.{SecurityManager, SparkConf, SparkException}
Expand All @@ -54,11 +55,13 @@ import org.apache.spark.deploy.yarn.security.YARNHadoopDelegationTokenManager
import org.apache.spark.internal.Logging
import org.apache.spark.internal.config._
import org.apache.spark.launcher.{LauncherBackend, SparkAppHandle, YarnCommandBuilderUtils}
import org.apache.spark.rpc.RpcEnv
import org.apache.spark.util.{CallerContext, Utils}

private[spark] class Client(
val args: ClientArguments,
val sparkConf: SparkConf)
val sparkConf: SparkConf,
val rpcEnv: RpcEnv)
extends Logging {

import Client._
Expand All @@ -69,6 +72,9 @@ private[spark] class Client(

private val isClusterMode = sparkConf.get("spark.submit.deployMode", "client") == "cluster"

private val isClientUnmanagedAMEnabled = sparkConf.get(YARN_UNMANAGED_AM) && !isClusterMode
private var amServiceStarted = false

@vanzin vanzin Jan 17, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you need this extra flag? Could you just check if appMaster != null?


// AM related configurations
private val amMemory = if (isClusterMode) {
sparkConf.get(DRIVER_MEMORY).toInt
Expand Down Expand Up @@ -297,7 +303,7 @@ private[spark] class Client(
"does not support it", e)
}
}

appContext.setUnmanagedAM(isClientUnmanagedAMEnabled)
appContext
}

Expand Down Expand Up @@ -660,7 +666,9 @@ private[spark] class Client(
// Clear the cache-related entries from the configuration to avoid them polluting the
// UI's environment page. This works for client mode; for cluster mode, this is handled
// by the AM.
CACHE_CONFIGS.foreach(sparkConf.remove)
if (!isClientUnmanagedAMEnabled) {

@vanzin vanzin Feb 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is this needed in the new mode?

@devaraj-kavali devaraj-kavali Feb 15, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is clearing the classpath entries and leading to this error in Executors.

Error: Could not find or load main class org.apache.spark.executor.CoarseGrainedExecutorBackend

@vanzin vanzin Dec 10, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is happening because you're starting the AM after these are removed from the conf. Should probably juggle things around or change how these are provided to the AM, since these configs are super noisy and shouldn't really show up in the UI.

CACHE_CONFIGS.foreach(sparkConf.remove)
}

localResources
}
Expand Down Expand Up @@ -1098,14 +1106,41 @@ private[spark] class Client(
if (returnOnRunning && state == YarnApplicationState.RUNNING) {
return createAppReport(report)
}

if (state == YarnApplicationState.ACCEPTED && isClientUnmanagedAMEnabled &&
!amServiceStarted && report.getAMRMToken != null) {
amServiceStarted = true
startApplicationMasterService(report)
}
lastState = state
}

// Never reached, but keeps compiler happy
throw new SparkException("While loop is depleted! This should never happen...")
}

private def startApplicationMasterService(report: ApplicationReport) = {

@vanzin vanzin Dec 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

: Unit =

But given you should be explicitly stopping the AM, this should probably return the AM itself.

// Add AMRMToken to establish connection between RM and AM
val token = report.getAMRMToken
val amRMToken: org.apache.hadoop.security.token.Token[AMRMTokenIdentifier] =

@vanzin vanzin Feb 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do you need to make this copy? Isn't the Token above enough?

@devaraj-kavali devaraj-kavali Feb 15, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

report.getAMRMToken gives org.apache.hadoop.yarn.api.records.Token type instance, but currentUGI.addToken expects org.apache.hadoop.security.token.Token type instance.

new org.apache.hadoop.security.token.Token[AMRMTokenIdentifier](
token.getIdentifier().array(), token.getPassword().array,
new Text(token.getKind()), new Text(token.getService()))
val currentUGI = UserGroupInformation.getCurrentUser
currentUGI.addToken(amRMToken)

// Start Application Service in a separate thread and continue with application monitoring
val amService = new Thread("Unmanaged Application Master Service") {
override def run(): Unit = {
val appMaster = new ApplicationMaster(
new ApplicationMasterArguments(Array.empty), sparkConf, hadoopConf)
appMaster.runUnmanaged(rpcEnv, report.getCurrentApplicationAttemptId,
new Path(appStagingBaseDir, getAppStagingDir(report.getApplicationId)))
}
}
amService.setDaemon(true)
amService.start()
}

private def formatReportDetails(report: ApplicationReport): String = {
val details = Seq[(String, String)](
("client token", getClientToken(report)),
Expand Down Expand Up @@ -1527,7 +1562,7 @@ private[spark] class YarnClusterApplication extends SparkApplication {
conf.remove("spark.jars")
conf.remove("spark.files")

new Client(new ClientArguments(args), conf).run()
new Client(new ClientArguments(args), conf, null).run()
}

}
Expand Down
5 changes: 3 additions & 2 deletions resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnRMClient.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,9 +79,10 @@ private[spark] class YarnRMClient extends Logging {
driverUrl: String,
driverRef: RpcEndpointRef,
securityMgr: SecurityManager,
localResources: Map[String, LocalResource]): YarnAllocator = {
localResources: Map[String, LocalResource],
appAttemptId: ApplicationAttemptId): YarnAllocator = {
require(registered, "Must register AM before creating allocator.")
new YarnAllocator(driverUrl, driverRef, conf, sparkConf, amClient, getAttemptId(), securityMgr,
new YarnAllocator(driverUrl, driverRef, conf, sparkConf, amClient, appAttemptId, securityMgr,
localResources, new SparkRackResolver())
}

Expand Down
Loading