目錄

Maven pom引用

配置文件

代碼

---

??? 平時開發(fā)中可以在yarn的web頁面查看應用程序運行狀態(tài)，如下圖

?

下面代碼實現(xiàn)了，代碼監(jiān)控Yarn運行程序，可以對部分任務進行實時監(jiān)控

Maven pom引用

這里Demo使用的hadoop版本是 3.0.0

<dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-yarn-api</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-yarn-client</artifactId><version>${hadoop.version}</version></dependency>

配置文件

yarn-site.xml 放在resources目錄下

代碼

import java.io.IOException; import java.util.ArrayList; import java.util.EnumSet; import java.util.List; import java.util.concurrent.TimeUnit;import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationReport; import org.apache.hadoop.yarn.api.records.YarnApplicationState; import org.apache.hadoop.yarn.client.api.YarnClient; import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.exceptions.YarnException; import org.apache.hadoop.yarn.util.ConverterUtils;public class YarnMonitor {/*** 獲取任務的applicationId* @return String* @param jobName* @return*/public static String getAppId(String jobName) {YarnClient client = YarnClient.createYarnClient();Configuration conf = new Configuration();client.init(conf);client.start();EnumSet<YarnApplicationState> appStates = EnumSet.noneOf(YarnApplicationState.class);if (appStates.isEmpty()) {appStates.add(YarnApplicationState.RUNNING);appStates.add(YarnApplicationState.ACCEPTED);appStates.add(YarnApplicationState.SUBMITTED);}List<ApplicationReport> appsReport = null;try {//返回EnumSet<YarnApplicationState>中個人任務狀態(tài)的所有任務appsReport = client.getApplications(appStates);} catch (YarnException | IOException e) {e.printStackTrace();}assert appsReport != null;for (ApplicationReport appReport : appsReport) {//獲取任務名String jn = appReport.getName();String applicationType = appReport.getApplicationType();if (jn.equals(jobName) && "Apache Flink".equals(applicationType)) {try {client.close();} catch (IOException e) {e.printStackTrace();}return appReport.getApplicationId().toString();}}try {client.close();} catch (IOException e) {e.printStackTrace();}return null;}/*** 根據(jù)任務的applicationId去獲取任務的狀態(tài)* @return YarnApplicationState* @param appId* @return*/public static YarnApplicationState getState(String appId) {YarnClient client = YarnClient.createYarnClient();Configuration conf = new Configuration();client.init(conf);client.start();ApplicationId applicationId = ApplicationId.fromString(appId);// ApplicationId appId = ConverterUtils.toApplicationId(appId);YarnApplicationState yarnApplicationState = null;try {ApplicationReport applicationReport = client.getApplicationReport(applicationId);yarnApplicationState = applicationReport.getYarnApplicationState();} catch (YarnException | IOException e) {e.printStackTrace();}try {client.close();} catch (IOException e) {e.printStackTrace();}return yarnApplicationState;}public static void main(String[] args) throws IOException, InterruptedException {while(true) {TimeUnit.SECONDS.sleep(3);boolean yarnIsContains = yarnIsContains("Spark Pi");System.out.println(yarnIsContains);}}/*** 判斷任務名為appName的任務，是否在yarn中運行，狀態(tài)為RUNNING* @return boolean* @param appName* @return*/public static boolean yarnIsContains(String appName) {Configuration conf = new YarnConfiguration();YarnClient yarnClient = YarnClient.createYarnClient();yarnClient.init(conf);yarnClient.start();boolean isContains = false;List<ApplicationReport> applications = new ArrayList<ApplicationReport>();try {//applications = yarnClient.getApplications(EnumSet.of(YarnApplicationState.RUNNING, YarnApplicationState.FINISHED));applications = yarnClient.getApplications(EnumSet.of(YarnApplicationState.RUNNING));for(ApplicationReport application:applications) {String name = application.getName();if(name.equals(appName)) {System.out.println("ApplicationId ============> "+application.getApplicationId());System.out.println("name ============> "+application.getName());System.out.println("queue ============> "+application.getQueue());System.out.println("queue ============> "+application.getUser());System.out.println(applications);isContains = true;}}/** if(applications.contains(appName)) {* System.out.println("ApplicationId ============> "+applications.get(0).* getApplicationId());* System.out.println("name ============> "+applications.get(0).getName());* System.out.println("queue ============> "+applications.get(0).getQueue());* System.out.println("queue ============> "+applications.get(0).getUser());* System.out.println(applications); }*/} catch (YarnException e) {e.printStackTrace();} catch (IOException e) {e.printStackTrace();} finally {yarnClient.stop();} return isContains;} }

Yarn的狀態(tài)有以下幾種，可根據(jù)個人需要進行使用，上面Demo我們是判斷任務名為 Spark Pi 的程序是否處于運行狀態(tài)，如果是則返回true。

總結

以上是生活随笔為你收集整理的Yarn 监控 - 监控任务运行状态 （包括Spark，MR 所有在Yarn中运行的任务）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。