簡(jiǎn)介

distributedShell是Yarn自帶的應用程序，和MR類(lèi)似，當前工具可以用來(lái)對Yarn進(jìn)行壓測。

使用示例

參考命令如下：

./bin/hadoop jar ./share/hadoop/yarn/hadoop-yarn-applications-distributedshell-3.4.1.jar \
 -jar ./share/hadoop/yarn/hadoop-yarn-applications-distributedshell-3.4.1.jar  -shell_command \
'/bin/date' -num_containers 5 

可以提交一個(gè)樣例作業(yè)到Yarn上面。

源碼閱讀

當前樣例的入口類(lèi)是org.apache.hadoop.yarn.applications.distributedshell.Client ，在pom文件里面默認定義了當前類(lèi)為主類(lèi)。所以在提交的時(shí)候可以不用指定主類(lèi)。

 <plugin>
        <artifactId>maven-jar-plugin</artifactId>
        <executions>
         <!-- 省略部分參數 -->
        </executions>
        <configuration>
           <archive>
             <manifest>
               <mainClass>org.apache.hadoop.yarn.applications.distributedshell.Client</mainClass>
             </manifest>
           </archive>
        </configuration>
      </plugin>

核心流程主要包含下面3個(gè)：

• 初始化CLient對象
• 初始化Client
• 提交作業(yè)到yarn

其中前面兩個(gè)主要在客戶(hù)端，第3個(gè)主要是在yarn上面。

客戶(hù)端提交核心代碼

初始化

初始化階段包括下面兩部分：

• 初始化Client對象，主要是創(chuàng )建Yarn的連接以及初始化支持的參數列表
• 初始化Client

下面是初始化Client對象的核心代碼。

Client(String appMasterMainClass, Configuration conf) {
    this.conf = conf;
    this.conf.setBoolean(
        YarnConfiguration.YARN_CLIENT_LOAD_RESOURCETYPES_FROM_SERVER, true);
    this.appMasterMainClass = appMasterMainClass;
    // 創(chuàng  )建和RM的連接
    yarnClient = YarnClient.createYarnClient();
    yarnClient.init(conf);
    opts = new Options();
    // 初始化支持的參數列表
    stopSignalReceived = new AtomicBoolean(false);
    isRunning = new AtomicBoolean(false);
  }

初始化Client，在初始化Client階段主要是讀取命令行參數。

// 初始化Client函數入口
boolean doRun = client.init(args);

運行作業(yè)

首先還是建立和Yarn服務(wù)端的連接，為作業(yè)提交做準備。

 isRunning.set(true);
 yarnClient.start();

在連接建立之后會(huì )查詢(xún)并且在控制臺打印Yarn服務(wù)端的一些信息。主要包含下面內容：

• 當前集群NM的個(gè)數，通過(guò)yarnClient.getYarnClusterMetrics() 查詢(xún)到并且顯示。
• 當前集群中運行中NM的詳細信息，通過(guò)yarnClient.getNodeReports(NodeState.RUNNING)查詢(xún)到。
• 當前任務(wù)提交的隊列的詳細信息，通過(guò)yarnClient.getQueueInfo(this.amQueue)查詢(xún)到。
• 當前集群的ACL信息，通過(guò)yarnClient.getQueueAclsInfo()查詢(xún)。
• 當前集群的ResourceProfile信息，通過(guò)yarnClient.getResourceProfiles()查詢(xún)。

在打印完集群信息之后才是作業(yè)提交的開(kāi)始。

提交作業(yè)之前，是需要先向RM申請AppId的。AppId可以通過(guò)YarnClientApplication app = yarnClient.createApplication();獲取。作業(yè)提交信息一般都在A(yíng)pplicationSubmissionContext里面，包含下面信息：

• AM申請資源的請求。通過(guò)appContext.setAMContainerResourceRequests(amResourceRequests);設置。

• AM的上下文信息：

  • 訪(fǎng)問(wèn)hdfs等所需要的token。當前token會(huì )伴隨著(zhù)整個(gè)作業(yè)，直到作業(yè)結束才會(huì )異步銷(xiāo)毀。
  • 需要本地話(huà)的文件。
  • AM或者Container所需要的環(huán)境變量。
  • AM的啟動(dòng)命令,AM啟動(dòng)的類(lèi)也是在這里指定的。類(lèi)似于 java運行jar或者某個(gè)主類(lèi)。

• App名稱(chēng)。通過(guò)appContext.setApplicationName(appName);設置。

• app tag信息。

• 資源標簽信息。

• 作業(yè)的優(yōu)先級。

• 作業(yè)提交的隊列信息。

• 日志聚合相關(guān)配置。主要是和日志歸集的Rolling模式有關(guān)系?？梢栽O置需要通過(guò)rolling的方式歸集哪些日志。通過(guò)appContext.setLogAggregationContext(logAggregationContext);設置。

作業(yè)真正提交的代碼只有一行：

yarnClient.submitApplication(appContext);

當前樣例做到了作業(yè)所需要的信息可配置。是一個(gè)比較適合開(kāi)發(fā)作業(yè)的樣例。

AM核心代碼

AM的核心代碼是在A(yíng)pplicationMaster.java里面的。在啟動(dòng)AM的時(shí)候會(huì )調用到當前函數的main函數。

在構造函數里面和init函數里面，主要是加載配置項以及命令行參數。真正運行的函數是run，核心在run函數里面，

首先需要創(chuàng )建和RM以及NM的連接。

amRMClient = AMRMClientAsync.createAMRMClientAsync(1000, allocListener);
amRMClient.init(conf);
amRMClient.start();

containerListener = createNMCallbackHandler();
nmClientAsync = new NMClientAsyncImpl(containerListener);
nmClientAsync.init(conf);
nmClientAsync.start();  
startTimelineClient(conf);

在A(yíng)M啟動(dòng)OK了第一件事就是需要去RM上面注冊，證明當前AM已經(jīng)啟動(dòng)完成了。

RegisterApplicationMasterResponse response = amRMClient
        .registerApplicationMaster(appMasterHostname, appMasterRpcPort,
            appMasterTrackingUrl, placementConstraintMap);

普通Container的申請是在A(yíng)M里面處理的，類(lèi)似下面代碼，下面代碼是異步申請的。

ContainerRequest containerAsk = setupContainerAskForRM();
amRMClient.addContainerRequest(containerAsk);

當Container申請好之后，可以通過(guò)下面代碼獲取，在樣例中觸發(fā)onContainerAllocated事件。

List<Container> allocated = response.getAllocatedContainers();
if (!allocated.isEmpty()) {
    handler.onContainersAllocated(allocated);
}

通過(guò)下面代碼啟動(dòng)Container.

ContainerLaunchContext ctx = ContainerLaunchContext.newInstance(
        localResources, myShellEnv, commands, null, allTokens.duplicate(),
          null, containerRetryContext); 
nmClientAsync.startContainerAsync(container, ctx);

在作業(yè)結束的時(shí)候，AM需要做下面事：

• 停止nmClient。
• 從RM上取消AppMaster
• 停止amClient。

nmClientAsync.stop();
try {
    amRMClient.unregisterApplicationMaster(appStatus, message, null);
} catch (YarnException | IOException ex) {
    LOG.error("Failed to unregister application", ex);
}
amRMClient.stop();