JobTracker节点后台线程之CompleteJobStatusStore

缺乏、安全感 2022-06-14 11:23 160阅读 0赞

上一篇博文讲了JobTracker节点会在内存中缓存一部分已完成Jobs状态信息，其实JobTracker节点还可以持久化地保存这些已完成Jobs信息，即将每一个Job的状态信息保存在一个单独的文件中，这个文件可以属于本地文件系统，也可以属于分布式文件系统，如：HDFS。所以，JobTracker节点设计了一个组件——CompleteJobStatusStore，来负责读/写和管理这些作业状态文件。对于任何一个作业，无论它最后是成功完成，还是失败，也不管它是否中途被JobTracke节点终止，JobTracker节点都会使用CompleteJobStatusStore来将这个已完成的Job状态信息保存到一个单独文件中。JobTracker节点之所以要持久化地保存这些Job的状态信息，一是因为这样做可以为用户提供查询Job的历史状态信息；二是因为内存有限，不可能缓存所有的Job状态信息。同时，出于系统资源的限制，JobTracker不可能永远的持久化保存一个Job的状态信息，以至于每一个已完成Job状态信息的持久化保存有一个生命周期，即JobTracker节点会定时的检测有哪些已完成Job持久化状态信息过期了，对于过期的，它会清除它们对应的文件。而这个工作也是交由它的后台线程CompleteJobStatusStore来完成的。

这里所说的Job的状态信息主要包括：JobStatus、JobProfile、Counters、TaskCompletionEvent，这些信息在文件中的组织形式如下：

![0_133155648481X5.gif][]

1.JobStatus(作业的基本状态信息)

private JobID jobid;//作业ID
    private float mapProgress;//map任务的最后执行进度
    private float reduceProgress;//reduce任务的最后执行进度
    private float cleanupProgress;//cleanup任务的最后执行进度
    private float setupProgress;//setup任务的最后执行进度
    private int runState;//作业最后结束时的状态
    private long startTime;//作业的开始时间
    private String user;//作业的提交者
    private JobPriority priority;//作业调度的优先级
    private String schedulingInfo="NA";//作业的调度信息

2.JobProfile(作业的基本配置信息)

String user;//提交作业的用户名
    final JobID jobid;//作业的ID
    String jobFile;//作业的配置文件job.xml的路径
    String url;//作业详细信息的http地址
    String name;//用户定义的作业名
    String queueName;//作业所属的队列

3.Counters(作业的过程统计器)

![0_1331558387oOyM.gif][]

4.TaskCompletionEvents(作业的任务完成事件)

private int eventId; //事件Id
    private String taskTrackerHttp;//运行该任务的TaskTracker节点的http服务地址
    private int taskRunTime; // 任务运行时间
    private TaskAttemptID taskId;//任务实例Id
    Status status; //任务结束状态
    boolean isMap = false;//任务类型
    private int idWithinJob;//任务在作业内部的Id

前面说过CompleteJobStatusStore不仅负责Job状态信息的读/写，还负责对这些文件进行管理，这个管理主要集中在每隔**SLEEP\_TIME** ms就开始检查一次，对于持久化保存时间超过retainTimeh的已完成Job状态信息保存文件要立即删除，这里的**SLEEP\_TIME**值为24\*60\*60\*1000，而retainTime的值可以通过配置文件来设置，对应的配置项为：mapred.job.tracker.persist.jobstatus.hours。另外，还有一个问题就是这些文件应该统一存放在那个目录下呢？关于这个存储目录，默认的是/jobtracker/JobsInfo目录，不过也可通过配置文件来设置，对应的配置项为：mapred.job.tracker.persist.jobstatus.dir。

CompleteJobStatusStore是否真的持久化地保存一个Job的状态信息，主要取决于它的两个参数：retainTime和active，也就是只有当retainTime>0 && active=true 时才会保存。这里的active起到了对进行开关的作用，是否开启CompleteJobStatusStore服务可以通过配置参数mapred.job.tracker.persist.jobstatus.active来设置，但默认是关闭的。

贴出这个循环检测的源代码：

public void run() {
        if (retainTime > 0) {
          while (true) {
            deleteJobStatusDirs();
            try {
              Thread.sleep(SLEEP_TIME);
            }
            catch (InterruptedException ex) {
              break;
            }
          }
        }
      }
    
      private void deleteJobStatusDirs() {
        try {
          long currentTime = System.currentTimeMillis();
          FileStatus[] jobInfoFiles = fs.listStatus(new Path[]{new Path(jobInfoDir)});
    
          //noinspection ForLoopReplaceableByForEach
          for (FileStatus jobInfo : jobInfoFiles) {
            try {
              if ((currentTime - jobInfo.getModificationTime()) > retainTime) {
                fs.delete(jobInfo.getPath(), true);
              }
            }
            catch (IOException ie) {
              LOG.warn("Could not do housekeeping for [ " + jobInfo.getPath() + "] job info : " + ie.getMessage(), ie);
            }
          }
        }
        catch (IOException ie) {
          LOG.warn("Could not obtain job info files : " + ie.getMessage(), ie);
        }
      }

[0_133155648481X5.gif]: /images/20220614/3a837373b04e44198760fc08bb5da7bb.png
[0_1331558387oOyM.gif]: /images/20220614/8a2259a12de44571b4cb443871898568.png