在用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 milliseconds

    public 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的api
        YarnClient yarnClient = YarnClient.createYarnClient();
        yarnClient.init(conf);
        yarnClient.start();

        // 调用Yarn的api，获取所有正在运行的应用程序
        List<ApplicationReport> 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();
    }
}

如果要监控oozie的调度任务，也可以用下面的方法：

import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

// 导入oozie的api相关的类
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.WorkflowJob;

public class OozieJobMonitor {

    // 定义一个正则表达式，用于匹配作业的运行时间
    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 milliseconds

    public static void main(String[] args) throws Exception {
        // 创建一个OozieClient对象，用于调用oozie的api
        OozieClient oozieClient = new OozieClient("http://localhost:11000/oozie");
        // 调用oozie的api，查询所有正在运行的作业
        List<WorkflowJob> jobs = oozieClient.getJobsInfo("status=RUNNING");
        // 遍历每个作业
        for (WorkflowJob job : jobs) {
            // 获取作业的ID和信息
            String jobId = job.getId();
            String jobInfo = job.toString();
            // 解析作业的信息，获取作业的运行时间
            long jobDuration = parseJobDuration(jobInfo);
            // 判断作业的运行时间是否超过阈值
            if (jobDuration > TIMEOUT_THRESHOLD) {
                // 如果超过阈值，调用oozie的api，终止作业
                oozieClient.kill(jobId);
                // 打印日志或者做其他操作
                System.out.println("Job " + jobId + " is killed due to timeout");
                // 重新运行作业或者做其他操作
                // ...
            } else {
                // 如果没有超过阈值，打印日志或者做其他操作
                System.out.println("Job " + jobId + " is running normally");
            }
        }
    }

    // 定义一个方法，用于解析作业的信息，并返回作业的运行时间（毫秒）
    private static long parseJobDuration(String jobInfo) {
        // 创建一个Matcher对象，用于匹配正则表达式和作业信息
        Matcher matcher = DURATION_PATTERN.matcher(jobInfo);
        // 如果找到了匹配的结果，就从结果中提取天数、小时数和分钟数，并转换为毫秒
        if (matcher.find()) {
            int days = Integer.parseInt(matcher.group(1));
            int hours = Integer.parseInt(matcher.group(2));
            int minutes = Integer.parseInt(matcher.group(3));
            long duration = (days * 24 + hours) * 60 + minutes; // duration in minutes
            duration *= 60 * 1000; // duration in milliseconds
            return duration;
        } else {
            // 如果没有找到匹配的结果，就返回-1表示无法解析
            return -1;
        }
    }
}

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典 ，回复“大数据”获取多本大数据电子书