@Override
  public MapReduceApplicationData fetchData(AnalyticJob job) throws IOException {
    DataFiles files = getHistoryFiles(job);
    String confFile = files.getJobConfPath();
    String histFile = files.getJobHistPath();
    String appId = job.getAppId();
    String jobId = Utils.getJobIdFromApplicationId(appId);

    MapReduceApplicationData jobData = new MapReduceApplicationData();
    jobData.setAppId(appId).setJobId(jobId);

    // Fetch job config
    Configuration jobConf = new Configuration(false);
    jobConf.addResource(_fs.open(new Path(confFile)), confFile);
    Properties jobConfProperties = new Properties();
    for (Map.Entry<String, String> entry : jobConf) {
      jobConfProperties.put(entry.getKey(), entry.getValue());
    }
    jobData.setJobConf(jobConfProperties);

    // Check if job history file is too large and should be throttled
    if (_fs.getFileStatus(new Path(histFile)).getLen() > _maxLogSizeInMB * FileUtils.ONE_MB) {
      String errMsg =
          "The history log of MapReduce application: " + appId + " is over the limit size of " + _maxLogSizeInMB + " MB, the parsing process gets throttled.";
      logger.warn(errMsg);
      jobData.setDiagnosticInfo(errMsg);
      jobData.setSucceeded(false);  // set succeeded to false to avoid heuristic analysis
      return jobData;
    }

    // Analyze job history file
    JobHistoryParser parser = new JobHistoryParser(_fs, histFile);
    JobHistoryParser.JobInfo jobInfo = parser.parse();
    IOException parseException = parser.getParseException();
    if (parseException != null) {
      throw new RuntimeException("Could not parse history file " + histFile, parseException);
    }
    // Populate missing fields from parsed job info. This info will be missing for backfilled jobs.
    populateJobFromJobInfo(job, jobInfo);

    jobData.setSubmitTime(jobInfo.getSubmitTime());
    jobData.setStartTime(jobInfo.getLaunchTime());
    jobData.setFinishTime(jobInfo.getFinishTime());

    String state = jobInfo.getJobStatus();
    if (state.equals("SUCCEEDED")) {
      jobData.setSucceeded(true);
    } else if (state.equals("FAILED")) {
      jobData.setSucceeded(false);
      jobData.setDiagnosticInfo(jobInfo.getErrorInfo());
    } else {
      throw new RuntimeException("job neither succeeded or failed. can not process it ");
    }

    // Fetch job counter
    MapReduceCounterData jobCounter = getCounterData(jobInfo.getTotalCounters());

    // Fetch task data
    Map<TaskID, JobHistoryParser.TaskInfo> allTasks = jobInfo.getAllTasks();
    List<JobHistoryParser.TaskInfo> mapperInfoList = new ArrayList<JobHistoryParser.TaskInfo>();
    List<JobHistoryParser.TaskInfo> reducerInfoList = new ArrayList<JobHistoryParser.TaskInfo>();
    for (JobHistoryParser.TaskInfo taskInfo : allTasks.values()) {
      if (taskInfo.getTaskType() == TaskType.MAP) {
        mapperInfoList.add(taskInfo);
      } else {
        reducerInfoList.add(taskInfo);
      }
    }
    if (jobInfo.getTotalMaps() > MAX_SAMPLE_SIZE) {
      logger.debug(jobId + " total mappers: " + mapperInfoList.size());
    }
    if (jobInfo.getTotalReduces() > MAX_SAMPLE_SIZE) {
      logger.debug(jobId + " total reducers: " + reducerInfoList.size());
    }
    MapReduceTaskData[] mapperList = getTaskData(jobId, mapperInfoList);
    MapReduceTaskData[] reducerList = getTaskData(jobId, reducerInfoList);

    jobData.setCounters(jobCounter).setMapperData(mapperList).setReducerData(reducerList);

    return jobData;
  }
The location of a job history file is in format: {done-dir}/yyyy/mm/dd/{serialPart}.
yyyy/mm/dd is the year, month and date of the finish time.
serialPart is the first 6 digits of the serial number considering it as a 9 digits number.
PS: The serial number is the last part of an app id.
<p>
For example, if appId = application_1461566847127_84624, then serial number is 84624.
Consider it as a 9 digits number, serial number is 000084624. So the corresponding
serialPart is 000084. If this application finish at 2016-5-30, its history file will locate
at {done-dir}/2016/05/30/000084
</p>
<p>
Furthermore, this location format is only satisfied for finished jobs in {done-dir} and not
for running jobs in {intermediate-done-dir}.
</p>
app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFSFetcherHadoop2.java