Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/3896#discussion_r22592495
--- Diff: yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala ---
@@ -105,25 +127,498 @@ private[spark] class Client(
}
/** Set up security tokens for launching our ApplicationMaster
container. */
- override def setupSecurityToken(amContainer: ContainerLaunchContext):
Unit = {
+ private def setupSecurityToken(amContainer: ContainerLaunchContext):
Unit = {
val dob = new DataOutputBuffer
credentials.writeTokenStorageToStream(dob)
amContainer.setTokens(ByteBuffer.wrap(dob.getData))
}
/** Get the application report from the ResourceManager for an
application we have submitted. */
- override def getApplicationReport(appId: ApplicationId):
ApplicationReport =
+ def getApplicationReport(appId: ApplicationId): ApplicationReport =
yarnClient.getApplicationReport(appId)
/**
* 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.
*/
- override def getClientToken(report: ApplicationReport): String =
+ private def getClientToken(report: ApplicationReport): String =
Option(report.getClientToAMToken).map(_.toString).getOrElse("")
+
+ /**
+ * Fail fast if we have requested more resources per container than is
available in the cluster.
+ */
+ private def verifyClusterResources(newAppResponse:
GetNewApplicationResponse): Unit = {
+ val maxMem = newAppResponse.getMaximumResourceCapability().getMemory()
+ logInfo("Verifying our application has not requested more than the
maximum " +
+ s"memory capability of the cluster ($maxMem MB per container)")
+ val executorMem = args.executorMemory + executorMemoryOverhead
+ if (executorMem > maxMem) {
+ throw new IllegalArgumentException(s"Required executor memory
(${args.executorMemory}" +
+ s"+$executorMemoryOverhead MB) is above the max threshold ($maxMem
MB) of this cluster!")
+ }
+ val amMem = args.amMemory + amMemoryOverhead
+ if (amMem > maxMem) {
+ throw new IllegalArgumentException(s"Required AM memory
(${args.amMemory}" +
+ s"+$amMemoryOverhead MB) is above the max threshold ($maxMem MB)
of this cluster!")
+ }
+ logInfo("Will allocate AM container, with %d MB memory including %d MB
overhead".format(
+ amMem,
+ amMemoryOverhead))
+
+ // We could add checks to make sure the entire cluster has enough
resources but that involves
+ // getting all the node reports and computing ourselves.
+ }
+
+ /**
+ * Copy the given file to a remote file system (e.g. HDFS) if needed.
+ * The file is only copied if the source and destination file systems
are different. This is used
+ * for preparing resources for launching the ApplicationMaster
container. Exposed for testing.
+ */
+ def copyFileToRemote(
--- End diff --
should these just go into `object Client` too? (How many extra parameters
do we need to pass them in?) If possible I'd prefer to keep the class itself
smaller
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
