/** * Process one {@link HistoryEvent} * * @param event * The {@link HistoryEvent} to be processed. */ public void process(HistoryEvent event) { if (event instanceof TaskAttemptFinishedEvent) { processTaskAttemptFinishedEvent((TaskAttemptFinishedEvent) event); } else if (event instanceof TaskAttemptUnsuccessfulCompletionEvent) { processTaskAttemptUnsuccessfulCompletionEvent((TaskAttemptUnsuccessfulCompletionEvent) event); } else if (event instanceof TaskStartedEvent) { processTaskStartedEvent((TaskStartedEvent) event); } else if (event instanceof MapAttemptFinishedEvent) { processMapAttemptFinishedEvent((MapAttemptFinishedEvent) event); } else if (event instanceof ReduceAttemptFinishedEvent) { processReduceAttemptFinishedEvent((ReduceAttemptFinishedEvent) event); } // I do NOT expect these if statements to be exhaustive. }
HistoryEvent maybeEmitEvent(ParsedLine line, String taskAttemptIDName, HistoryEventEmitter thatg) { if (taskAttemptIDName == null) { return null; } TaskAttemptID taskAttemptID = TaskAttemptID.forName(taskAttemptIDName); String finishTime = line.get("FINISH_TIME"); String status = line.get("TASK_STATUS"); if (finishTime != null && status != null && status.equalsIgnoreCase("success")) { String hostName = line.get("HOSTNAME"); String counters = line.get("COUNTERS"); String state = line.get("STATE_STRING"); TaskAttempt20LineEventEmitter that = (TaskAttempt20LineEventEmitter) thatg; ParsedHost pHost = ParsedHost.parse(hostName); return new TaskAttemptFinishedEvent(taskAttemptID, that.originalTaskType, status, Long.parseLong(finishTime), pHost.getRackName(), pHost.getNodeName(), state, maybeParseCounters(counters)); } return null; }
private void processTaskAttemptFinishedEvent(TaskAttemptFinishedEvent event) { ParsedTaskAttempt attempt = getOrMakeTaskAttempt(event.getTaskType(), event.getTaskId().toString(), event.getAttemptId().toString()); if (attempt == null) { return; } attempt.setResult(getPre21Value(event.getTaskStatus())); ParsedHost pHost = getAndRecordParsedHost(event.getRackName(), event.getHostname()); if (pHost != null) { attempt.setLocation(pHost.makeLoggedLocation()); } attempt.setFinishTime(event.getFinishTime()); attempt .incorporateCounters(((TaskAttemptFinished) event.getDatum()).counters); }
HistoryEvent maybeEmitEvent(ParsedLine line, String taskAttemptIDName, HistoryEventEmitter thatg) { if (taskAttemptIDName == null) { return null; } TaskAttemptID taskAttemptID = TaskAttemptID.forName(taskAttemptIDName); String finishTime = line.get("FINISH_TIME"); String status = line.get("TASK_STATUS"); if (finishTime != null && status != null && status.equalsIgnoreCase("success")) { String hostName = line.get("RPC_ADDRESSES"); String counters = line.get("COUNTERS"); String state = line.get("STATE_STRING"); TaskAttempt20LineEventEmitter that = (TaskAttempt20LineEventEmitter) thatg; ParsedHost pHost = ParsedHost.parse(hostName); return new TaskAttemptFinishedEvent(taskAttemptID, that.originalTaskType, status, Long.parseLong(finishTime), pHost.getRackName(), pHost.getNodeName(), state, maybeParseCounters(counters)); } return null; }
/** * Process one {@link HistoryEvent} * * @param event * The {@link HistoryEvent} to be processed. */ public void process(HistoryEvent event) { if (event instanceof TaskAttemptFinishedEvent) { processTaskAttemptFinishedEvent((TaskAttemptFinishedEvent) event); } else if (event instanceof TaskAttemptUnsuccessfulCompletionEvent) { processTaskAttemptUnsuccessfulCompletionEvent((TaskAttemptUnsuccessfulCompletionEvent) event); } else if (event instanceof TaskStartedEvent) { processTaskStartedEvent((TaskStartedEvent) event); } // I do NOT expect these if statements to be exhaustive. }
HistoryEvent maybeEmitEvent(ParsedLine line, String taskAttemptIDName, HistoryEventEmitter thatg) { if (taskAttemptIDName == null) { return null; } TaskAttemptID taskAttemptID = TaskAttemptID.forName(taskAttemptIDName); String finishTime = line.get("FINISH_TIME"); String status = line.get("TASK_STATUS"); if (finishTime != null && status != null && status.equalsIgnoreCase("success")) { String hostName = line.get("HOSTNAME"); String counters = line.get("COUNTERS"); String state = line.get("STATE_STRING"); TaskAttempt20LineEventEmitter that = (TaskAttempt20LineEventEmitter) thatg; return new TaskAttemptFinishedEvent(taskAttemptID, that.originalTaskType, status, Long.parseLong(finishTime), hostName, state, maybeParseCounters(counters)); } return null; }
private void processTaskAttemptFinishedEvent(TaskAttemptFinishedEvent event) { LoggedTaskAttempt attempt = getOrMakeTaskAttempt(event.getTaskType(), event.getTaskId().toString(), event.getAttemptId().toString()); if (attempt == null) { return; } attempt.setResult(getPre21Value(event.getTaskStatus())); attempt.setLocation(getAndRecordParsedHost(event.getHostname()) .makeLoggedLocation()); attempt.setFinishTime(event.getFinishTime()); attempt .incorporateCounters(((TaskAttemptFinished) event.getDatum()).counters); }
private void processTaskAttemptFinishedEvent(TaskAttemptFinishedEvent event) { recordParsedHost(event.getHostname(), event.getRackName()); }
/** * Process one {@link HistoryEvent} * * @param event * The {@link HistoryEvent} to be processed. */ public void process(HistoryEvent event) { if (finalized) { throw new IllegalStateException( "JobBuilder.process(HistoryEvent event) called after ParsedJob built"); } // these are in lexicographical order by class name. if (event instanceof AMStartedEvent) { // ignore this event as Rumen currently doesnt need this event //TODO Enhance Rumen to process this event and capture restarts return; } else if (event instanceof NormalizedResourceEvent) { // Log an warn message as NormalizedResourceEvent shouldn't be written. LOG.warn("NormalizedResourceEvent should be ignored in history server."); } else if (event instanceof JobFinishedEvent) { processJobFinishedEvent((JobFinishedEvent) event); } else if (event instanceof JobInfoChangeEvent) { processJobInfoChangeEvent((JobInfoChangeEvent) event); } else if (event instanceof JobInitedEvent) { processJobInitedEvent((JobInitedEvent) event); } else if (event instanceof JobPriorityChangeEvent) { processJobPriorityChangeEvent((JobPriorityChangeEvent) event); } else if (event instanceof JobQueueChangeEvent) { processJobQueueChangeEvent((JobQueueChangeEvent) event); } else if (event instanceof JobStatusChangedEvent) { processJobStatusChangedEvent((JobStatusChangedEvent) event); } else if (event instanceof JobSubmittedEvent) { processJobSubmittedEvent((JobSubmittedEvent) event); } else if (event instanceof JobUnsuccessfulCompletionEvent) { processJobUnsuccessfulCompletionEvent((JobUnsuccessfulCompletionEvent) event); } else if (event instanceof MapAttemptFinishedEvent) { processMapAttemptFinishedEvent((MapAttemptFinishedEvent) event); } else if (event instanceof ReduceAttemptFinishedEvent) { processReduceAttemptFinishedEvent((ReduceAttemptFinishedEvent) event); } else if (event instanceof TaskAttemptFinishedEvent) { processTaskAttemptFinishedEvent((TaskAttemptFinishedEvent) event); } else if (event instanceof TaskAttemptStartedEvent) { processTaskAttemptStartedEvent((TaskAttemptStartedEvent) event); } else if (event instanceof TaskAttemptUnsuccessfulCompletionEvent) { processTaskAttemptUnsuccessfulCompletionEvent((TaskAttemptUnsuccessfulCompletionEvent) event); } else if (event instanceof TaskFailedEvent) { processTaskFailedEvent((TaskFailedEvent) event); } else if (event instanceof TaskFinishedEvent) { processTaskFinishedEvent((TaskFinishedEvent) event); } else if (event instanceof TaskStartedEvent) { processTaskStartedEvent((TaskStartedEvent) event); } else if (event instanceof TaskUpdatedEvent) { processTaskUpdatedEvent((TaskUpdatedEvent) event); } else throw new IllegalArgumentException( "JobBuilder.process(HistoryEvent): unknown event type:" + event.getEventType() + " for event:" + event); }
/** * Process one {@link HistoryEvent} * * @param event * The {@link HistoryEvent} to be processed. */ public void process(HistoryEvent event) { if (finalized) { throw new IllegalStateException( "JobBuilder.process(HistoryEvent event) called after ParsedJob built"); } // these are in lexicographical order by class name. if (event instanceof AMStartedEvent) { // ignore this event as Rumen currently doesnt need this event //TODO Enhance Rumen to process this event and capture restarts return; } else if (event instanceof JobFinishedEvent) { processJobFinishedEvent((JobFinishedEvent) event); } else if (event instanceof JobInfoChangeEvent) { processJobInfoChangeEvent((JobInfoChangeEvent) event); } else if (event instanceof JobInitedEvent) { processJobInitedEvent((JobInitedEvent) event); } else if (event instanceof JobPriorityChangeEvent) { processJobPriorityChangeEvent((JobPriorityChangeEvent) event); } else if (event instanceof JobStatusChangedEvent) { processJobStatusChangedEvent((JobStatusChangedEvent) event); } else if (event instanceof JobSubmittedEvent) { processJobSubmittedEvent((JobSubmittedEvent) event); } else if (event instanceof JobUnsuccessfulCompletionEvent) { processJobUnsuccessfulCompletionEvent((JobUnsuccessfulCompletionEvent) event); } else if (event instanceof MapAttemptFinishedEvent) { processMapAttemptFinishedEvent((MapAttemptFinishedEvent) event); } else if (event instanceof ReduceAttemptFinishedEvent) { processReduceAttemptFinishedEvent((ReduceAttemptFinishedEvent) event); } else if (event instanceof TaskAttemptFinishedEvent) { processTaskAttemptFinishedEvent((TaskAttemptFinishedEvent) event); } else if (event instanceof TaskAttemptStartedEvent) { processTaskAttemptStartedEvent((TaskAttemptStartedEvent) event); } else if (event instanceof TaskAttemptUnsuccessfulCompletionEvent) { processTaskAttemptUnsuccessfulCompletionEvent((TaskAttemptUnsuccessfulCompletionEvent) event); } else if (event instanceof TaskFailedEvent) { processTaskFailedEvent((TaskFailedEvent) event); } else if (event instanceof TaskFinishedEvent) { processTaskFinishedEvent((TaskFinishedEvent) event); } else if (event instanceof TaskStartedEvent) { processTaskStartedEvent((TaskStartedEvent) event); } else if (event instanceof TaskUpdatedEvent) { processTaskUpdatedEvent((TaskUpdatedEvent) event); } else throw new IllegalArgumentException( "JobBuilder.process(HistoryEvent): unknown event type"); }