|
|
@@ -91,15 +91,6 @@ class Client(yarnConf: YarnConfiguration) {
|
|
|
val resource = Records.newRecord(classOf[Resource])
|
|
|
|
|
|
info("Copy app master jar from local filesystem and add to the local environment")
|
|
|
- /*
|
|
|
- val localResources = Map(
|
|
|
- "app" => uploadLocalResource()
|
|
|
- )
|
|
|
- Map
|
|
|
-
|
|
|
-
|
|
|
- addToLocalResources(fs, appMasterJar, appMasterJarPath, appId, localResources, null)
|
|
|
- */
|
|
|
|
|
|
val packageResource = Records.newRecord(classOf[LocalResource])
|
|
|
|
|
|
@@ -225,208 +216,3 @@ case class Accepted() extends ApplicationStatus
|
|
|
case class Running() extends ApplicationStatus
|
|
|
case class SuccessfulFinish() extends ApplicationStatus
|
|
|
case class UnsuccessfulFinish() extends ApplicationStatus
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-/*
|
|
|
-object Client {
|
|
|
-
|
|
|
- def main(args: Array[String]) = {
|
|
|
- val jarPath = new Path(args(1))
|
|
|
-
|
|
|
- val yarnConf = new YarnConfiguration()
|
|
|
- val yarnClient = YarnClient.createYarnClient()
|
|
|
-
|
|
|
- yarnClient.init(yarnConf)
|
|
|
- yarnClient.start()
|
|
|
-
|
|
|
- try {
|
|
|
- val app = yarnClient.createApplication()
|
|
|
- val amContainer: ContainerLaunchContext = Records.newRecord(Class[ContainerLaunchContext])
|
|
|
- amContainer.setCommands(
|
|
|
- Collections.singletonList(
|
|
|
- "$JAVA_HOME/bin/java" +
|
|
|
- " com.cloudera.hue.livy.repl.yarn.ApplicationMaster" +
|
|
|
- " 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout" +
|
|
|
- " 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout"))
|
|
|
-
|
|
|
- val appMasterJar: LocalResource = Records.newRecord(Class[LocalResource])
|
|
|
- setupAppMasterJar(jarPath, appMasterJar)
|
|
|
- amContainer.setLocalResources(
|
|
|
- Collections.singletonMap("foo.jar", appMasterJar)
|
|
|
- )
|
|
|
-
|
|
|
- val appMasterEnv: Map[String, String] = Map()
|
|
|
- setupAppMasterEnv(appMasterEnv)
|
|
|
- amContainer.setEnvironment(appMasterEnv)
|
|
|
-
|
|
|
- val capability: Resource = Records.newRecord(Class[Resource])
|
|
|
- capability.setMemory(256)
|
|
|
- capability.setVirtualCores(1)
|
|
|
-
|
|
|
- val appContext = app.getApplicationSubmissionContext
|
|
|
- appContext.setApplicationName("foo")
|
|
|
- appContext.setAMContainerSpec(amContainer)
|
|
|
- appContext.setResource(capability)
|
|
|
- appContext.setQueue("default")
|
|
|
-
|
|
|
- val appId = appContext.getApplicationId
|
|
|
- yarnClient.submitApplication(appContext)
|
|
|
-
|
|
|
- var appReport = yarnClient.getApplicationReport(appId)
|
|
|
- var appState = appReport.getYarnApplicationState()
|
|
|
-
|
|
|
- while (
|
|
|
- appState != YarnApplicationState.FINISHED &&
|
|
|
- appState != YarnApplicationState.KILLED &&
|
|
|
- appState != YarnApplicationState.FAILED
|
|
|
- ) {
|
|
|
- Thread.sleep(100)
|
|
|
- appReport = yarnClient.getApplicationReport(appId)
|
|
|
- appState = appReport.getYarnApplicationState
|
|
|
- }
|
|
|
-
|
|
|
- } finally {
|
|
|
- yarnClient.close()
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- private def setupAppMasterJar(value: Path, resource: LocalResource) = {
|
|
|
-
|
|
|
- }
|
|
|
-
|
|
|
- private def setupAppMasterEnv(conf: YarnConfiguration, appMasterEnv: Map[String, String]) = {
|
|
|
- var classpaths = conf.getStrings(YarnConfiguration.YARN_APPLICATION_CLASSPATH)
|
|
|
-
|
|
|
- if (classpaths == null) {
|
|
|
- classpaths = YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH
|
|
|
- }
|
|
|
-
|
|
|
- classpaths.foreach {
|
|
|
- c => {
|
|
|
- Apps.addToEnvironment(appMasterEnv, Environment.CLASSPATH.name(), c.trim())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- Apps.addToEnvironment(appMasterEnv, Environment.CLASSPATH.name(),
|
|
|
- Environment.PWD.$() + File.separator + "*"
|
|
|
- )
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- /*
|
|
|
-
|
|
|
- try {
|
|
|
- val appContext = yarnClient.createApplication.getApplicationSubmissionContext
|
|
|
- val appId = appContext.getApplicationId
|
|
|
-
|
|
|
- val appName = "livy-repl"
|
|
|
- val amPriority = 0
|
|
|
- val amQueue = "default"
|
|
|
-
|
|
|
- appContext.setApplicationName(appName)
|
|
|
-
|
|
|
- val priority: Priority = Records.newRecord(Class[Priority])
|
|
|
- priority.setPriority(amPriority)
|
|
|
- appContext.setPriority(priority)
|
|
|
-
|
|
|
- appContext.setQueue(amQueue)
|
|
|
-
|
|
|
- val amContainer: ContainerLaunchContext = Records.newRecord(Class[ContainerLaunchContext])
|
|
|
- appContext.setAMContainerSpec(amContainer)
|
|
|
-
|
|
|
- appContext.setUnmanagedAM(true)
|
|
|
-
|
|
|
- yarnClient.submitApplication(appContext)
|
|
|
-
|
|
|
- var appReport = monitorApplication(
|
|
|
- appId,
|
|
|
- util.EnumSet.of(
|
|
|
- YarnApplicationState.ACCEPTED,
|
|
|
- YarnApplicationState.KILLED,
|
|
|
- YarnApplicationState.FAILED,
|
|
|
- YarnApplicationState.FINISHED
|
|
|
- ))
|
|
|
-
|
|
|
- if (appReport.getYarnApplicationState == YarnApplicationState.ACCEPTED) {
|
|
|
- val attemptReport = monitorCurrentAppAttempt(appId, YarnApplicationAttemptState.LAUNCHED)
|
|
|
- val attemptId = attemptReport.getApplicationAttemptId
|
|
|
-
|
|
|
- launchAM(yarnClient, attemptId)
|
|
|
-
|
|
|
- appReport = monitorApplication(
|
|
|
- appId,
|
|
|
- util.EnumSet.of(
|
|
|
- YarnApplicationState.KILLED,
|
|
|
- YarnApplicationState.FAILED,
|
|
|
- YarnApplicationState.FINISHED
|
|
|
- )
|
|
|
- )
|
|
|
- }
|
|
|
-
|
|
|
- val appState = appReport.getYarnApplicationState
|
|
|
- val appStatus = appReport.getFinalApplicationStatus
|
|
|
-
|
|
|
- if (YarnApplicationState.FINISHED == appState && FinalApplicationStatus.SUCCEEDED == appStatus) {
|
|
|
- 0
|
|
|
- } else {
|
|
|
- 1
|
|
|
- }
|
|
|
- } finally {
|
|
|
- yarnClient.close()
|
|
|
- }
|
|
|
- }
|
|
|
- */
|
|
|
-
|
|
|
- /*
|
|
|
- private def launchAM(rmClient: YarnClient, attemptId: ApplicationAttemptId): Unit = {
|
|
|
- val credentials = new Credentials();
|
|
|
- val token = rmClient.getAMRMToken(attemptId.getApplicationId)
|
|
|
- credentials.addToken(token.getService, token)
|
|
|
- val tokenFile = File.createTempFile("unmanagedAMRMToken", "", new File(System.getProperty("user.dir")));
|
|
|
- //try {
|
|
|
- FileUtil.chmod(tokenFile.getAbsolutePath, "600")
|
|
|
- //}
|
|
|
-
|
|
|
- tokenFile.deleteOnExit()
|
|
|
- val os = new DataOutputStream(new FileOutputStream(tokenFile, true))
|
|
|
- credentials.writeTokenStorageToStream(os)
|
|
|
- os.close()
|
|
|
-
|
|
|
- val envAMList = List()
|
|
|
- var setClasspath = false
|
|
|
- val classpath = null
|
|
|
-
|
|
|
- sys.env.foreach {
|
|
|
- case(key, value) => {
|
|
|
- var value: String = value
|
|
|
- if (key == "CLASSPATH") {
|
|
|
- setClasspath = true
|
|
|
- if (classpath != null) {
|
|
|
- value = value + File.pathSeparator + classpath
|
|
|
- }
|
|
|
- }
|
|
|
- envAMList +: (key + "=" + value)
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if (!setClasspath && classpath != null) {
|
|
|
- envAMList +: ("CLASSPATH=" + classpath)
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- }
|
|
|
-
|
|
|
- private def monitorApplication(appId: ApplicationId, attemptState: util.EnumSet[YarnApplicationState]): ApplicationReport = {
|
|
|
- null
|
|
|
- }
|
|
|
-
|
|
|
- private def monitorCurrentAppAttempt(appId: ApplicationId, attemptState: YarnApplicationAttemptState): ApplicationAttemptReport = {
|
|
|
- null
|
|
|
- }
|
|
|
- */
|
|
|
-
|
|
|
-
|
|
|
-}
|
|
|
-*/
|