org.apache.hadoop.mapreduce.TaskType#JOB_CLEANUP源码实例Demo

/**
 * @param type Task type.
 * @return Hadoop task type.
 */
private TaskType taskType(HadoopTaskType type) {
    switch (type) {
        case SETUP:
            return TaskType.JOB_SETUP;
        case MAP:
        case COMBINE:
            return TaskType.MAP;

        case REDUCE:
            return TaskType.REDUCE;

        case COMMIT:
        case ABORT:
            return TaskType.JOB_CLEANUP;

        default:
            return null;
    }
}
 

static TaskType get20TaskType(String taskType) {
  try {
    return TaskType.valueOf(taskType);
  } catch (IllegalArgumentException e) {
    if ("CLEANUP".equals(taskType)) {
      return TaskType.JOB_CLEANUP;
    }

    if ("SETUP".equals(taskType)) {
      return TaskType.JOB_SETUP;
    }

    return null;
  }
}
 

static TaskType get20TaskType(String taskType) {
  try {
    return TaskType.valueOf(taskType);
  } catch (IllegalArgumentException e) {
    if ("CLEANUP".equals(taskType)) {
      return TaskType.JOB_CLEANUP;
    }

    if ("SETUP".equals(taskType)) {
      return TaskType.JOB_SETUP;
    }

    return null;
  }
}
 

/**
 * Returns the type of the {@link TaskAttemptID} passed. 
 * The type of an attempt is determined by the nature of the task and not its 
 * id. 
 * For example,
 * - Attempt 'attempt_123_01_m_01_0' might be a job-setup task even though it 
 *   has a _m_ in its id. Hence the task type of this attempt is JOB_SETUP 
 *   instead of MAP.
 * - Similarly reduce attempt 'attempt_123_01_r_01_0' might have failed and is
 *   now supposed to do the task-level cleanup. In such a case this attempt 
 *   will be of type TASK_CLEANUP instead of REDUCE.
 */
TaskType getAttemptType (TaskAttemptID id) {
  if (isCleanupAttempt(id)) {
    return TaskType.TASK_CLEANUP;
  } else if (isJobSetupTask()) {
    return TaskType.JOB_SETUP;
  } else if (isJobCleanupTask()) {
    return TaskType.JOB_CLEANUP;
  } else if (isMapTask()) {
    return TaskType.MAP;
  } else {
    return TaskType.REDUCE;
  }
}
 

private TaskInProgress createAndAddTIP(JobTracker jobtracker, 
                                       JobInProgress jip, TaskType type) {
  JobConf conf = jip.getJobConf();
  JobID id = jip.getJobID();
  // now create a fake tip for this fake job
  TaskInProgress tip = null;
  if (type == TaskType.MAP) {
    tip = new TaskInProgress(id, "dummy", new JobClient.RawSplit(), 
                             conf, jip, 0, 1);
    jip.maps = new TaskInProgress[] {tip};
  } else if (type == TaskType.REDUCE) {
    tip = new TaskInProgress(id, "dummy", jip.desiredMaps(), 0, 
                             conf, jip, 1);
    jip.reduces = new TaskInProgress[] {tip};
  } else if (type == TaskType.JOB_SETUP) {
    tip = 
      new TaskInProgress(id, "dummy", new JobClient.RawSplit(), 
                         conf, jip, 0, 1);
    jip.setup = new TaskInProgress[] {tip};
  } else if (type == TaskType.JOB_CLEANUP) {
    tip = 
      new TaskInProgress(id, "dummy", new JobClient.RawSplit(), 
                         conf, jip, 0, 1);
    jip.cleanup = new TaskInProgress[] {tip};
  }
  return tip;
}
 

/**
 * Creates cleanup {@link TaskAttemptContext} for given {@link JobID}.
 *
 * @param conf hadoop configuration
 * @param jobID jobId of the created {@link TaskID}
 * @return new cleanup {@link TaskID} for given {@link JobID}
 */
static TaskAttemptContext createCleanupTaskContext(Configuration conf, JobID jobID) {
  final TaskID taskId = new TaskID(jobID, TaskType.JOB_CLEANUP, 0);
  return createTaskAttemptContext(conf, new TaskAttemptID(taskId, 0));
}