org.apache.hadoop.mapreduce.TaskAttemptID#toString ( )源码实例Demo

/**
 * Create an event to record the start of an attempt
 * @param attemptId Id of the attempt
 * @param taskType Type of task
 * @param startTime Start time of the attempt
 * @param trackerName Name of the Task Tracker where attempt is running
 * @param httpPort The port number of the tracker
 * @param shufflePort The shuffle port number of the container
 * @param containerId The containerId for the task attempt.
 * @param locality The locality of the task attempt
 * @param avataar The avataar of the task attempt
 */
public TaskAttemptStartedEvent( TaskAttemptID attemptId,  
    TaskType taskType, long startTime, String trackerName,
    int httpPort, int shufflePort, ContainerId containerId,
    String locality, String avataar) {
  datum.attemptId = new Utf8(attemptId.toString());
  datum.taskid = new Utf8(attemptId.getTaskID().toString());
  datum.startTime = startTime;
  datum.taskType = new Utf8(taskType.name());
  datum.trackerName = new Utf8(trackerName);
  datum.httpPort = httpPort;
  datum.shufflePort = shufflePort;
  datum.containerId = new Utf8(containerId.toString());
  if (locality != null) {
    datum.locality = new Utf8(locality);
  }
  if (avataar != null) {
    datum.avataar = new Utf8(avataar);
  }
}
 

/**
 * Create an event to record the start of an attempt
 * @param attemptId Id of the attempt
 * @param taskType Type of task
 * @param startTime Start time of the attempt
 * @param trackerName Name of the Task Tracker where attempt is running
 * @param httpPort The port number of the tracker
 * @param shufflePort The shuffle port number of the container
 * @param containerId The containerId for the task attempt.
 * @param locality The locality of the task attempt
 * @param avataar The avataar of the task attempt
 */
public TaskAttemptStartedEvent( TaskAttemptID attemptId,  
    TaskType taskType, long startTime, String trackerName,
    int httpPort, int shufflePort, ContainerId containerId,
    String locality, String avataar) {
  datum.attemptId = new Utf8(attemptId.toString());
  datum.taskid = new Utf8(attemptId.getTaskID().toString());
  datum.startTime = startTime;
  datum.taskType = new Utf8(taskType.name());
  datum.trackerName = new Utf8(trackerName);
  datum.httpPort = httpPort;
  datum.shufflePort = shufflePort;
  datum.containerId = new Utf8(containerId.toString());
  if (locality != null) {
    datum.locality = new Utf8(locality);
  }
  if (avataar != null) {
    datum.avataar = new Utf8(avataar);
  }
}
 

@Override
public void cleanUpPartialOutputForTask(TaskAttemptContext context)
    throws IOException {

  // we double check this is never invoked from a non-preemptable subclass.
  // This should never happen, since the invoking codes is checking it too,
  // but it is safer to double check. Errors handling this would produce
  // inconsistent output.

  if (!this.getClass().isAnnotationPresent(Checkpointable.class)) {
    throw new IllegalStateException("Invoking cleanUpPartialOutputForTask() " +
        "from non @Preemptable class");
  }
  FileSystem fs =
    fsFor(getTaskAttemptPath(context), context.getConfiguration());

  LOG.info("cleanUpPartialOutputForTask: removing everything belonging to " +
      context.getTaskAttemptID().getTaskID() + " in: " +
      getCommittedTaskPath(context).getParent());

  final TaskAttemptID taid = context.getTaskAttemptID();
  final TaskID tid = taid.getTaskID();
  Path pCommit = getCommittedTaskPath(context).getParent();
  // remove any committed output
  for (int i = 0; i < taid.getId(); ++i) {
    TaskAttemptID oldId = new TaskAttemptID(tid, i);
    Path pTask = new Path(pCommit, oldId.toString());
    if (fs.exists(pTask) && !fs.delete(pTask, true)) {
      throw new IOException("Failed to delete " + pTask);
    }
  }
}
 

@Override
public void cleanUpPartialOutputForTask(TaskAttemptContext context)
    throws IOException {

  // we double check this is never invoked from a non-preemptable subclass.
  // This should never happen, since the invoking codes is checking it too,
  // but it is safer to double check. Errors handling this would produce
  // inconsistent output.

  if (!this.getClass().isAnnotationPresent(Checkpointable.class)) {
    throw new IllegalStateException("Invoking cleanUpPartialOutputForTask() " +
        "from non @Preemptable class");
  }
  FileSystem fs =
    fsFor(getTaskAttemptPath(context), context.getConfiguration());

  LOG.info("cleanUpPartialOutputForTask: removing everything belonging to " +
      context.getTaskAttemptID().getTaskID() + " in: " +
      getCommittedTaskPath(context).getParent());

  final TaskAttemptID taid = context.getTaskAttemptID();
  final TaskID tid = taid.getTaskID();
  Path pCommit = getCommittedTaskPath(context).getParent();
  // remove any committed output
  for (int i = 0; i < taid.getId(); ++i) {
    TaskAttemptID oldId = new TaskAttemptID(tid, i);
    Path pTask = new Path(pCommit, oldId.toString());
    if (fs.exists(pTask) && !fs.delete(pTask, true)) {
      throw new IOException("Failed to delete " + pTask);
    }
  }
}
 

protected MapReduceTaskData[] getTaskData(String jobId, List<JobHistoryParser.TaskInfo> infoList) {
  int sampleSize = sampleAndGetSize(jobId, infoList);

  List<MapReduceTaskData> taskList = new ArrayList<MapReduceTaskData>();
  for (int i = 0; i < sampleSize; i++) {
    JobHistoryParser.TaskInfo tInfo = infoList.get(i);

    String taskId = tInfo.getTaskId().toString();
    TaskAttemptID attemptId = null;
    if(tInfo.getTaskStatus().equals("SUCCEEDED")) {
      attemptId = tInfo.getSuccessfulAttemptId();
    } else {
      attemptId = tInfo.getFailedDueToAttemptId();
    }

    MapReduceTaskData taskData = new MapReduceTaskData(taskId, attemptId == null ? "" : attemptId.toString() , tInfo.getTaskStatus());

    MapReduceCounterData taskCounterData = getCounterData(tInfo.getCounters());

    long[] taskExecTime = null;
    if (attemptId != null) {
      taskExecTime = getTaskExecTime(tInfo.getAllTaskAttempts().get(attemptId));
    }

    taskData.setTimeAndCounter(taskExecTime, taskCounterData);
    taskList.add(taskData);
  }
  return taskList.toArray(new MapReduceTaskData[taskList.size()]);
}
 

private String getAttemptString(Reducer<Text, NullWritable, Text, BooleanWritable>.Context context) {
  TaskAttemptID taskAttemptID = context.getTaskAttemptID();
  return taskAttemptID.toString();
}