云南网站建设天锐科技,国外app素材网站,centos7 wordpress,网站开发文献综述在用oozie的调度任务#xff0c;用shell调度spark任务#xff0c;在生产环境运行时#xff0c;正常1-2个小时跑完的任务#xff0c;有时出现跑了5、6个小时还没跑完#xff0c;造成的原因很奇怪#xff0c;有可能是数据倾斜#xff0c;任务占用太多资源偶尔出错。为了监…在用oozie的调度任务用shell调度spark任务在生产环境运行时正常1-2个小时跑完的任务有时出现跑了5、6个小时还没跑完造成的原因很奇怪有可能是数据倾斜任务占用太多资源偶尔出错。为了监控这种现象并设定阈值为3个小时如果超过3小时没跑完就kill掉。可以结合oozie失败重试机制实现重跑。
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;// 导入oozie的api相关的类
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.client.api.YarnClient;public class YarnJobMonitor {// 定义一个正则表达式用于匹配作业的运行时间private static final Pattern DURATION_PATTERN Pattern.compile(Duration\\s*:\\s*(\\d) days, (\\d) hours, (\\d) minutes);// 定义一个常量表示超时的阈值3小时private static final long TIMEOUT_THRESHOLD 3 * 60 * 60 * 1000; // 3 hours in millisecondspublic static void main(String[] args) throws Exception {// 创建一个Configuration对象用于加载Hadoop和Yarn的配置文件Configuration conf new Configuration();conf.addResource(core-site.xml);conf.addResource(hdfs-site.xml);conf.addResource(yarn-site.xml);// 创建一个YarnClient对象用于访问Yarn的apiYarnClient yarnClient YarnClient.createYarnClient();yarnClient.init(conf);yarnClient.start();// 调用Yarn的api获取所有正在运行的应用程序ListApplicationReport apps yarnClient.getApplications(EnumSet.of(YarnApplicationState.RUNNING));// 遍历每个应用程序for (ApplicationReport app : apps) {// 获取应用程序的ID和名称ApplicationId appId app.getApplicationId();String appName app.getName();// 判断应用程序是否是由Oozie Shell命令启动的spark任务if (appName.startsWith(oozie:launcher)) {// 如果是打印日志或者做其他操作System.out.println(Found Oozie Shell spark job: appId);// 获取应用程序的开始时间和当前时间long startTime app.getStartTime();long currentTime System.currentTimeMillis();// 计算应用程序的运行时间毫秒long jobDuration currentTime - startTime;// 判断应用程序的运行时间是否超过阈值if (jobDuration TIMEOUT_THRESHOLD) {// 如果超过阈值调用Yarn的api终止应用程序yarnClient.killApplication(appId);// 打印日志或者做其他操作System.out.println(Killed Oozie Shell spark job: appId);// 重新运行应用程序或者做其他操作// ...} else {// 如果没有超过阈值打印日志或者做其他操作System.out.println(Job appId is running normally);}}}// 关闭YarnClient对象yarnClient.stop();}
}
