[SPARK-3477] Clean up code in Yarn Client / ClientBase #2350
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -23,154 +23,127 @@ import org.apache.hadoop.conf.Configuration | |
| import org.apache.hadoop.io.DataOutputBuffer | ||
| import org.apache.hadoop.security.UserGroupInformation | ||
| import org.apache.hadoop.yarn.api._ | ||
| import org.apache.hadoop.yarn.api.protocolrecords._ | ||
| import org.apache.hadoop.yarn.api.records._ | ||
| import org.apache.hadoop.yarn.client.YarnClientImpl | ||
| import org.apache.hadoop.yarn.conf.YarnConfiguration | ||
| import org.apache.hadoop.yarn.util.Records | ||
|
|
||
| import org.apache.spark.{Logging, SparkConf} | ||
| import org.apache.spark.deploy.SparkHadoopUtil | ||
|
|
||
| /** | ||
| * Version of [[org.apache.spark.deploy.yarn.ClientBase]] tailored to YARN's alpha API. | ||
| */ | ||
| private[spark] class Client( | ||
| val args: ClientArguments, | ||
| val hadoopConf: Configuration, | ||
| val sparkConf: SparkConf) | ||
| extends YarnClientImpl with ClientBase with Logging { | ||
|
|
||
| def this(clientArgs: ClientArguments, spConf: SparkConf) = | ||
| this(clientArgs, SparkHadoopUtil.get.newConfiguration(spConf), spConf) | ||
|
|
||
| def this(clientArgs: ClientArguments) = this(clientArgs, new SparkConf()) | ||
|
|
||
| val yarnConf: YarnConfiguration = new YarnConfiguration(hadoopConf) | ||
|
|
||
| /* ------------------------------------------------------------------------------------- * | ||
| | The following methods have much in common in the stable and alpha versions of Client, | | ||
| | but cannot be implemented in the parent trait due to subtle API differences across | | ||
| | hadoop versions. | | ||
| * ------------------------------------------------------------------------------------- */ | ||
|
|
||
| /** Submit an application running our ApplicationMaster to the ResourceManager. */ | ||
| override def submitApplication(): ApplicationId = { | ||
| init(yarnConf) | ||
| start() | ||
|
|
||
| logInfo("Requesting a new application from cluster with %d NodeManagers" | ||
| .format(getYarnClusterMetrics.getNumNodeManagers)) | ||
|
|
||
| // Get a new application from our RM | ||
| val newAppResponse = getNewApplication() | ||
| val appId = newAppResponse.getApplicationId() | ||
|
|
||
| // Verify whether the cluster has enough resources for our AM | ||
| verifyClusterResources(newAppResponse) | ||
|
|
||
| // Set up the appropriate contexts to launch our AM | ||
| val containerContext = createContainerLaunchContext(newAppResponse) | ||
| val appContext = createApplicationSubmissionContext(appId, containerContext) | ||
|
|
||
| // Finally, submit and monitor the application | ||
| logInfo(s"Submitting application ${appId.getId} to ResourceManager") | ||
| submitApplication(appContext) | ||
| appId | ||
| } | ||
|
|
||
| /** | ||
| * Set up a context for launching our ApplicationMaster container. | ||
| * In the Yarn alpha API, the memory requirements of this container must be set in | ||
| * the ContainerLaunchContext instead of the ApplicationSubmissionContext. | ||
| */ | ||
| override def createContainerLaunchContext(newAppResponse: GetNewApplicationResponse) | ||
| : ContainerLaunchContext = { | ||
| val containerContext = super.createContainerLaunchContext(newAppResponse) | ||
| val capability = Records.newRecord(classOf[Resource]) | ||
| capability.setMemory(args.amMemory + amMemoryOverhead) | ||
| containerContext.setResource(capability) | ||
| containerContext | ||
| } | ||
|
|
||
| /** Set up the context for submitting our ApplicationMaster. */ | ||
| def createApplicationSubmissionContext( | ||
| appId: ApplicationId, | ||
| containerContext: ContainerLaunchContext): ApplicationSubmissionContext = { | ||
| val appContext = Records.newRecord(classOf[ApplicationSubmissionContext]) | ||
| appContext.setApplicationId(appId) | ||
| appContext.setApplicationName(args.appName) | ||
| appContext.setQueue(args.amQueue) | ||
| appContext.setAMContainerSpec(containerContext) | ||
| appContext.setUser(UserGroupInformation.getCurrentUser.getShortUserName) | ||
| appContext | ||
| } | ||
|
|
||
| /** | ||
| * Set up security tokens for launching our ApplicationMaster container. | ||
| * ContainerLaunchContext#setContainerTokens is renamed `setTokens` in the stable API. | ||
| */ | ||
| override def setupSecurityToken(amContainer: ContainerLaunchContext): Unit = { | ||
| val dob = new DataOutputBuffer() | ||
| credentials.writeTokenStorageToStream(dob) | ||
| amContainer.setContainerTokens(ByteBuffer.wrap(dob.getData())) | ||
| } | ||
|
|
||
| /** | ||
| * Return the security token used by this client to communicate with the ApplicationMaster. | ||
| * If no security is enabled, the token returned by the report is null. | ||
|
There was a problem hiding this comment. I verified that this statement is true for the stable API. @tgravescs do you know if this is true for alpha? (Do we need a special null check here) There was a problem hiding this comment. yes getClientToken can return null and does if security is off. |
||
| * ApplicationReport#getClientToken is renamed `getClientToAMToken` in the stable API. | ||
| */ | ||
| override def getClientToken(report: ApplicationReport): String = | ||
| Option(report.getClientToken).getOrElse("") | ||
| } | ||
|
|
||
| object Client { | ||
| def main(argStrings: Array[String]) { | ||
| if (!sys.props.contains("SPARK_SUBMIT")) { | ||
| println("WARNING: This client is deprecated and will be removed in a " + | ||
| "future version of Spark. Use ./bin/spark-submit with \"--master yarn\"") | ||
| } | ||
|
|
||
| // Set an env variable indicating we are running in YARN mode. | ||
| // Note that any env variable with the SPARK_ prefix gets propagated to all (remote) processes | ||
| System.setProperty("SPARK_YARN_MODE", "true") | ||
| val sparkConf = new SparkConf | ||
|
|
||
| try { | ||
| val args = new ClientArguments(argStrings, sparkConf) | ||
| new Client(args, sparkConf).run() | ||
| } catch { | ||
| case e: Exception => | ||
| Console.err.println(e.getMessage) | ||
| System.exit(1) | ||
| } | ||
|
|
||
| System.exit(0) | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -17,15 +17,14 @@ | |
|
|
||
| package org.apache.spark.deploy.yarn | ||
|
|
||
| import scala.collection.mutable.ArrayBuffer | ||
|
|
||
| import org.apache.spark.SparkConf | ||
| import org.apache.spark.util.{Utils, IntParam, MemoryParam} | ||
|
|
||
|
|
||
| // TODO: Add code and support for ensuring that yarn resource 'tasks' are location aware ! | ||
| private[spark] class ClientArguments(args: Array[String], sparkConf: SparkConf) { | ||
| var addJars: String = null | ||
| var files: String = null | ||
| var archives: String = null | ||
|
|
@@ -35,28 +34,56 @@ class ClientArguments(val args: Array[String], val sparkConf: SparkConf) { | |
| var executorMemory = 1024 // MB | ||
| var executorCores = 1 | ||
| var numExecutors = 2 | ||
| var amQueue = sparkConf.get("spark.yarn.queue", "default") | ||
| var amMemory: Int = 512 // MB | ||
| var appName: String = "Spark" | ||
| var priority = 0 | ||
|
|
||
| // Additional memory to allocate to containers | ||
| // For now, use driver's memory overhead as our AM container's memory overhead | ||
| val amMemoryOverhead = sparkConf.getInt( | ||
| "spark.yarn.driver.memoryOverhead", YarnSparkHadoopUtil.DEFAULT_MEMORY_OVERHEAD) | ||
| val executorMemoryOverhead = sparkConf.getInt( | ||
| "spark.yarn.executor.memoryOverhead", YarnSparkHadoopUtil.DEFAULT_MEMORY_OVERHEAD) | ||
|
|
||
| parseArgs(args.toList) | ||
| loadEnvironmentArgs() | ||
| validateArgs() | ||
|
|
||
| /** Load any default arguments provided through environment variables and Spark properties. */ | ||
| private def loadEnvironmentArgs(): Unit = { | ||
| // For backward compatibility, SPARK_YARN_DIST_{ARCHIVES/FILES} should be resolved to hdfs://, | ||
| // while spark.yarn.dist.{archives/files} should be resolved to file:// (SPARK-2051). | ||
| files = Option(files) | ||
| .orElse(sys.env.get("SPARK_YARN_DIST_FILES")) | ||
| .orElse(sparkConf.getOption("spark.yarn.dist.files").map(p => Utils.resolveURIs(p))) | ||
| .orNull | ||
| archives = Option(archives) | ||
| .orElse(sys.env.get("SPARK_YARN_DIST_ARCHIVES")) | ||
| .orElse(sparkConf.getOption("spark.yarn.dist.archives").map(p => Utils.resolveURIs(p))) | ||
| .orNull | ||
| } | ||
|
|
||
| /** | ||
| * Fail fast if any arguments provided are invalid. | ||
| * This is intended to be called only after the provided arguments have been parsed. | ||
| */ | ||
| private def validateArgs(): Unit = { | ||
| // TODO: memory checks are outdated (SPARK-3476) | ||
| Map[Boolean, String]( | ||
| (numExecutors <= 0) -> "You must specify at least 1 executor!", | ||
| (amMemory <= amMemoryOverhead) -> s"AM memory must be > $amMemoryOverhead MB", | ||
| (executorMemory <= executorMemoryOverhead) -> | ||
| s"Executor memory must be > $executorMemoryOverhead MB" | ||
|
There was a problem hiding this comment. @tgravescs You had a few comments about a few of these checks being outdated in a separate PR. Which ones are no longer relevant? There was a problem hiding this comment. the memory ones are. see https://issues.apache.org/jira/browse/SPARK-3476 There was a problem hiding this comment. I would say just leave it as in for this pr and we can handle under the jira I filed. There was a problem hiding this comment. Alright, I'll leave these as is for now. There was a problem hiding this comment. I removed them here: #2528 |
||
| ).foreach { case (errorCondition, errorMessage) => | ||
| if (errorCondition) { | ||
| throw new IllegalArgumentException(errorMessage + "\n" + getUsageMessage()) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private def parseArgs(inputArgs: List[String]): Unit = { | ||
| val userArgsBuffer = new ArrayBuffer[String]() | ||
| var args = inputArgs | ||
|
|
||
| while (!args.isEmpty) { | ||
|
|
@@ -138,16 +165,14 @@ class ClientArguments(val args: Array[String], val sparkConf: SparkConf) { | |
| userArgs = userArgsBuffer.readOnly | ||
| } | ||
|
|
||
| private def getUsageMessage(unknownParam: List[String] = null): String = { | ||
| val message = if (unknownParam != null) s"Unknown/unsupported param $unknownParam\n" else "" | ||
| message + | ||
| "Usage: org.apache.spark.deploy.yarn.Client [options] \n" + | ||
| "Options:\n" + | ||
| " --jar JAR_PATH Path to your application's JAR file (required in yarn-cluster mode)\n" + | ||
| " --class CLASS_NAME Name of your application's main class (required)\n" + | ||
| " --arg ARG Argument to be passed to your application's main class.\n" + | ||
| " Multiple invocations are possible, each will be passed in order.\n" + | ||
| " --num-executors NUM Number of executors to start (Default: 2)\n" + | ||
| " --executor-cores NUM Number of cores for the executors (Default: 1).\n" + | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
In the past people might have actually used this directly to. I've seen people request in the past how to run the client from another program without going out to shell so this might break backwards compatibility also. We should probably file a jira to look at supporting that officially.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is addressed in the main discussion thread.