protected boolean launchDMLYarnAppmaster() throws IOException, DMLScriptException {
  boolean ret=false;
  String hdfsWD=null;
  try {
    Timing time=new Timing(true);
    YarnConfiguration yconf=new YarnConfiguration();
    YarnClient yarnClient=YarnClient.createYarnClient();
    yarnClient.init(yconf);
    yarnClient.start();
    YarnClientApplication app=yarnClient.createApplication();
    ApplicationSubmissionContext appContext=app.getApplicationSubmissionContext();
    ApplicationId appId=appContext.getApplicationId();
    LOG.debug("Created application (applicationID: " + appId + ")");
    hdfsWD=DMLAppMasterUtils.constructHDFSWorkingDir(_dmlConfig,appId);
    copyResourcesToHdfsWorkingDir(yconf,hdfsWD);
    String command=constructAMCommand(_args,_dmlConfig);
    LOG.debug("Constructed application master command: \n" + command);
    ContainerLaunchContext amContainer=Records.newRecord(ContainerLaunchContext.class);
    amContainer.setCommands(Collections.singletonList(command));
    amContainer.setLocalResources(constructLocalResourceMap(yconf));
    amContainer.setEnvironment(constructEnvionmentMap(yconf));
    int memHeap=_dmlConfig.getIntValue(DMLConfig.YARN_APPMASTERMEM);
    int memAlloc=(int)computeMemoryAllocation(memHeap);
    Resource capability=Records.newRecord(Resource.class);
    capability.setMemory(memAlloc);
    capability.setVirtualCores(NUM_CORES);
    LOG.debug("Requested application resources: memory=" + memAlloc + ", vcores="+ NUM_CORES);
    String qname=_dmlConfig.getTextValue(DMLConfig.YARN_APPQUEUE);
    appContext.setApplicationName(APPMASTER_NAME);
    appContext.setAMContainerSpec(amContainer);
    appContext.setResource(capability);
    appContext.setQueue(qname);
    LOG.debug("Configured application meta data: name=" + APPMASTER_NAME + ", queue="+ qname);
    yarnClient.submitApplication(appContext);
    ApplicationReport appReport=yarnClient.getApplicationReport(appId);
    LOG.info("Application tracking-URL: " + appReport.getTrackingUrl());
    YarnApplicationState appState=appReport.getYarnApplicationState();
    YarnApplicationState oldState=appState;
    LOG.info("Application state: " + appState);
    while (appState != YarnApplicationState.FINISHED && appState != YarnApplicationState.KILLED && appState != YarnApplicationState.FAILED) {
      Thread.sleep(APP_STATE_INTERVAL);
      appReport=yarnClient.getApplicationReport(appId);
      appState=appReport.getYarnApplicationState();
      if (appState != oldState) {
        oldState=appState;
        LOG.info("Application state: " + appState);
      }
    }
    FinalApplicationStatus finalState=appReport.getFinalApplicationStatus();
    LOG.info("Application final status: " + finalState);
    double appRuntime=(double)(appReport.getFinishTime() - appReport.getStartTime()) / 1000;
    LOG.info("Application runtime: " + appRuntime + " sec.");
    LOG.info("Total runtime: " + String.format("%.3f",time.stop() / 1000) + " sec.");
    if (finalState != FinalApplicationStatus.SUCCEEDED) {
      String stop_msg=readMessageToHDFSWorkingDir(_dmlConfig,yconf,appId);
      if (stop_msg != null)       throw new DMLScriptException(stop_msg);
      throw new DMLRuntimeException("DML yarn app master finished with final status: " + finalState + ".");
    }
    ret=true;
  }
 catch (  DMLScriptException ex) {
    throw ex;
  }
catch (  Exception ex) {
    LOG.error("Failed to run DML yarn app master.",ex);
    ret=false;
  }
 finally {
    if (hdfsWD != null)     MapReduceTool.deleteFileIfExistOnHDFS(hdfsWD);
  }
  return ret;
}

protected boolean launchDMLYarnAppmaster ( ) throws IOException , DMLScriptException { boolean ret = false ; String hdfsWD = null ; try { Timing time = new Timing ( true ) ; YarnConfiguration yconf = new YarnConfiguration ( ) ; YarnClient yarnClient = YarnClient . createYarnClient ( ) ; yarnClient . init ( yconf ) ; yarnClient . start ( ) ; YarnClientApplication app = yarnClient . createApplication ( ) ; ApplicationSubmissionContext appContext = app . getApplicationSubmissionContext ( ) ; ApplicationId appId = appContext . getApplicationId ( ) ; LOG . debug ( "Created application (applicationID: " + appId + ")" ) ; hdfsWD = DMLAppMasterUtils . constructHDFSWorkingDir ( _dmlConfig , appId ) ; copyResourcesToHdfsWorkingDir ( yconf , hdfsWD ) ; String command = constructAMCommand ( _args , _dmlConfig ) ; LOG . debug ( "Constructed application master command: \n" + command ) ; ContainerLaunchContext amContainer = Records . newRecord ( ContainerLaunchContext . class ) ; amContainer . setCommands ( Collections . singletonList ( command ) ) ; amContainer . setLocalResources ( constructLocalResourceMap ( yconf ) ) ; amContainer . setEnvironment ( constructEnvionmentMap ( yconf ) ) ; int memHeap = _dmlConfig . getIntValue ( DMLConfig . YARN_APPMASTERMEM ) ; int memAlloc = ( int ) computeMemoryAllocation ( memHeap ) ; Resource capability = Records . newRecord ( Resource . class ) ; capability . setMemory ( memAlloc ) ; capability . setVirtualCores ( NUM_CORES ) ; LOG . debug ( "Requested application resources: memory=" + memAlloc + ", vcores=" + NUM_CORES ) ; String qname = _dmlConfig . getTextValue ( DMLConfig . YARN_APPQUEUE ) ; appContext . setApplicationName ( APPMASTER_NAME ) ; appContext . setAMContainerSpec ( amContainer ) ; appContext . setResource ( capability ) ; appContext . setQueue ( qname ) ; LOG . debug ( "Configured application meta data: name=" + APPMASTER_NAME + ", queue=" + qname ) ; yarnClient . submitApplication ( appContext ) ; ApplicationReport appReport = yarnClient . getApplicationReport ( appId ) ; LOG . info ( "Application tracking-URL: " + appReport . getTrackingUrl ( ) ) ; YarnApplicationState appState = appReport . getYarnApplicationState ( ) ; YarnApplicationState oldState = appState ; LOG . info ( "Application state: " + appState ) ; while ( appState != YarnApplicationState . FINISHED && appState != YarnApplicationState . KILLED && appState != YarnApplicationState . FAILED ) { Thread . sleep ( APP_STATE_INTERVAL ) ; appReport = yarnClient . getApplicationReport ( appId ) ; appState = appReport . getYarnApplicationState ( ) ; if ( appState != oldState ) { oldState = appState ; LOG . info ( "Application state: " + appState ) ; } } FinalApplicationStatus finalState = appReport . getFinalApplicationStatus ( ) ; LOG . info ( "Application final status: " + finalState ) ; double appRuntime = ( double ) ( appReport . getFinishTime ( ) - appReport . getStartTime ( ) ) / 1000 ; LOG . info ( "Application runtime: " + appRuntime + " sec." ) ; LOG . info ( "Total runtime: " + String . format ( "%.3f" , time . stop ( ) / 1000 ) + " sec." ) ; if ( finalState != FinalApplicationStatus . SUCCEEDED ) { String stop_msg = readMessageToHDFSWorkingDir ( _dmlConfig , yconf , appId ) ; if ( stop_msg != null ) throw new DMLScriptException ( stop_msg ) ; throw new DMLRuntimeException ( "DML yarn app master finished with final status: " + finalState + "." ) ; } ret = true ; } catch ( DMLScriptException ex ) { throw ex ; } catch ( Exception ex ) { LOG . error ( "Failed to run DML yarn app master." , ex ) ; ret = false ; } finally { if ( hdfsWD != null ) MapReduceTool . deleteFileIfExistOnHDFS ( hdfsWD ) ; } return ret ; }
method to launch the dml yarn app master and execute the given dml script with the given configuration and jar file . note : on launching the yarn app master , we do not explicitly probe if we are running on a yarn or mr1 cluster . in case of mr1 , already the class yarnconfiguration will not be found and raise a classnotfound . in case of any exception we fall back to run cp directly in the client process .
