| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.hadoop.mapreduce.jobhistory; |
| |
| import java.io.IOException; |
| import java.text.DecimalFormat; |
| import java.text.Format; |
| import java.text.SimpleDateFormat; |
| import java.util.Arrays; |
| import java.util.Comparator; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.TreeSet; |
| |
| import org.apache.hadoop.classification.InterfaceAudience; |
| import org.apache.hadoop.classification.InterfaceStability; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.mapred.JobStatus; |
| import org.apache.hadoop.mapred.TaskStatus; |
| import org.apache.hadoop.mapreduce.CounterGroup; |
| import org.apache.hadoop.mapreduce.Counters; |
| import org.apache.hadoop.mapreduce.TaskAttemptID; |
| import org.apache.hadoop.mapreduce.TaskID; |
| import org.apache.hadoop.mapreduce.TaskType; |
| import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.JobInfo; |
| import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.TaskInfo; |
| import org.apache.hadoop.mapreduce.util.HostUtil; |
| import org.apache.hadoop.util.StringUtils; |
| |
| /** |
| * HistoryViewer is used to parse and view the JobHistory files |
| * |
| */ |
| @InterfaceAudience.Private |
| @InterfaceStability.Unstable |
| public class HistoryViewer { |
| private static SimpleDateFormat dateFormat = |
| new SimpleDateFormat("d-MMM-yyyy HH:mm:ss"); |
| private FileSystem fs; |
| private JobInfo job; |
| private String jobId; |
| private boolean printAll; |
| |
| /** |
| * Constructs the HistoryViewer object |
| * @param historyFile The fully qualified Path of the History File |
| * @param conf The Configuration file |
| * @param printAll Toggle to print all status to only killed/failed status |
| * @throws IOException |
| */ |
| public HistoryViewer(String historyFile, |
| Configuration conf, |
| boolean printAll) throws IOException { |
| this.printAll = printAll; |
| String errorMsg = "Unable to initialize History Viewer"; |
| try { |
| Path jobFile = new Path(historyFile); |
| fs = jobFile.getFileSystem(conf); |
| String[] jobDetails = |
| jobFile.getName().split("_"); |
| if (jobDetails.length < 2) { |
| // NOT a valid name |
| System.err.println("Ignore unrecognized file: " + jobFile.getName()); |
| throw new IOException(errorMsg); |
| } |
| JobHistoryParser parser = new JobHistoryParser(fs, jobFile); |
| job = parser.parse(); |
| jobId = job.getJobId().toString(); |
| } catch(Exception e) { |
| throw new IOException(errorMsg, e); |
| } |
| } |
| |
| /** |
| * Print the job/task/attempt summary information |
| * @throws IOException |
| */ |
| public void print() throws IOException{ |
| printJobDetails(); |
| printTaskSummary(); |
| printJobAnalysis(); |
| printTasks(TaskType.JOB_SETUP, TaskStatus.State.FAILED.toString()); |
| printTasks(TaskType.JOB_SETUP, TaskStatus.State.KILLED.toString()); |
| printTasks(TaskType.MAP, TaskStatus.State.FAILED.toString()); |
| printTasks(TaskType.MAP, TaskStatus.State.KILLED.toString()); |
| printTasks(TaskType.REDUCE, TaskStatus.State.FAILED.toString()); |
| printTasks(TaskType.REDUCE, TaskStatus.State.KILLED.toString()); |
| printTasks(TaskType.JOB_CLEANUP, TaskStatus.State.FAILED.toString()); |
| printTasks(TaskType.JOB_CLEANUP, |
| JobStatus.getJobRunState(JobStatus.KILLED)); |
| if (printAll) { |
| printTasks(TaskType.JOB_SETUP, TaskStatus.State.SUCCEEDED.toString()); |
| printTasks(TaskType.MAP, TaskStatus.State.SUCCEEDED.toString()); |
| printTasks(TaskType.REDUCE, TaskStatus.State.SUCCEEDED.toString()); |
| printTasks(TaskType.JOB_CLEANUP, TaskStatus.State.SUCCEEDED.toString()); |
| printAllTaskAttempts(TaskType.JOB_SETUP); |
| printAllTaskAttempts(TaskType.MAP); |
| printAllTaskAttempts(TaskType.REDUCE); |
| printAllTaskAttempts(TaskType.JOB_CLEANUP); |
| } |
| |
| FilteredJob filter = new FilteredJob(job, |
| TaskStatus.State.FAILED.toString()); |
| printFailedAttempts(filter); |
| |
| filter = new FilteredJob(job, |
| TaskStatus.State.KILLED.toString()); |
| printFailedAttempts(filter); |
| } |
| |
| private void printJobDetails() { |
| StringBuffer jobDetails = new StringBuffer(); |
| jobDetails.append("\nHadoop job: " ).append(job.getJobId()); |
| jobDetails.append("\n====================================="); |
| jobDetails.append("\nUser: ").append(job.getUsername()); |
| jobDetails.append("\nJobName: ").append(job.getJobname()); |
| jobDetails.append("\nJobConf: ").append(job.getJobConfPath()); |
| jobDetails.append("\nSubmitted At: ").append(StringUtils. |
| getFormattedTimeWithDiff(dateFormat, |
| job.getSubmitTime(), 0)); |
| jobDetails.append("\nLaunched At: ").append(StringUtils. |
| getFormattedTimeWithDiff(dateFormat, |
| job.getLaunchTime(), |
| job.getSubmitTime())); |
| jobDetails.append("\nFinished At: ").append(StringUtils. |
| getFormattedTimeWithDiff(dateFormat, |
| job.getFinishTime(), |
| job.getLaunchTime())); |
| jobDetails.append("\nStatus: ").append(((job.getJobStatus() == null) ? |
| "Incomplete" :job.getJobStatus())); |
| printCounters(jobDetails, job.getTotalCounters(), job.getMapCounters(), |
| job.getReduceCounters()); |
| jobDetails.append("\n"); |
| jobDetails.append("\n====================================="); |
| System.out.println(jobDetails.toString()); |
| } |
| |
| private void printCounters(StringBuffer buff, Counters totalCounters, |
| Counters mapCounters, Counters reduceCounters) { |
| // Killed jobs might not have counters |
| if (totalCounters == null) { |
| return; |
| } |
| buff.append("\nCounters: \n\n"); |
| buff.append(String.format("|%1$-30s|%2$-30s|%3$-10s|%4$-10s|%5$-10s|", |
| "Group Name", |
| "Counter name", |
| "Map Value", |
| "Reduce Value", |
| "Total Value")); |
| buff.append("\n------------------------------------------"+ |
| "---------------------------------------------"); |
| for (String groupName : totalCounters.getGroupNames()) { |
| CounterGroup totalGroup = totalCounters.getGroup(groupName); |
| CounterGroup mapGroup = mapCounters.getGroup(groupName); |
| CounterGroup reduceGroup = reduceCounters.getGroup(groupName); |
| |
| Format decimal = new DecimalFormat(); |
| Iterator<org.apache.hadoop.mapreduce.Counter> ctrItr = |
| totalGroup.iterator(); |
| while(ctrItr.hasNext()) { |
| org.apache.hadoop.mapreduce.Counter counter = ctrItr.next(); |
| String name = counter.getName(); |
| String mapValue = |
| decimal.format(mapGroup.findCounter(name).getValue()); |
| String reduceValue = |
| decimal.format(reduceGroup.findCounter(name).getValue()); |
| String totalValue = |
| decimal.format(counter.getValue()); |
| |
| buff.append( |
| String.format("%n|%1$-30s|%2$-30s|%3$-10s|%4$-10s|%5$-10s", |
| totalGroup.getDisplayName(), |
| counter.getDisplayName(), |
| mapValue, reduceValue, totalValue)); |
| } |
| } |
| } |
| |
| private void printAllTaskAttempts(TaskType taskType) { |
| Map<TaskID, TaskInfo> tasks = job.getAllTasks(); |
| StringBuffer taskList = new StringBuffer(); |
| taskList.append("\n").append(taskType); |
| taskList.append(" task list for ").append(job.getJobId()); |
| taskList.append("\nTaskId\t\tStartTime"); |
| if (TaskType.REDUCE.equals(taskType)) { |
| taskList.append("\tShuffleFinished\tSortFinished"); |
| } |
| taskList.append("\tFinishTime\tHostName\tError\tTaskLogs"); |
| taskList.append("\n===================================================="); |
| System.out.println(taskList.toString()); |
| for (JobHistoryParser.TaskInfo task : tasks.values()) { |
| for (JobHistoryParser.TaskAttemptInfo attempt : |
| task.getAllTaskAttempts().values()) { |
| if (taskType.equals(task.getTaskType())){ |
| taskList.setLength(0); |
| taskList.append(attempt.getAttemptId()).append("\t"); |
| taskList.append(StringUtils.getFormattedTimeWithDiff(dateFormat, |
| attempt.getStartTime(), 0)).append("\t"); |
| if (TaskType.REDUCE.equals(taskType)) { |
| taskList.append(StringUtils.getFormattedTimeWithDiff(dateFormat, |
| attempt.getShuffleFinishTime(), |
| attempt.getStartTime())); |
| taskList.append("\t"); |
| taskList.append(StringUtils.getFormattedTimeWithDiff(dateFormat, |
| attempt.getSortFinishTime(), |
| attempt.getShuffleFinishTime())); |
| } |
| taskList.append(StringUtils.getFormattedTimeWithDiff(dateFormat, |
| attempt.getFinishTime(), |
| attempt.getStartTime())); |
| taskList.append("\t"); |
| taskList.append(attempt.getHostname()).append("\t"); |
| taskList.append(attempt.getError()); |
| String taskLogsUrl = getTaskLogsUrl(attempt); |
| taskList.append(taskLogsUrl != null ? taskLogsUrl : "n/a"); |
| System.out.println(taskList.toString()); |
| } |
| } |
| } |
| } |
| |
| private void printTaskSummary() { |
| SummarizedJob ts = new SummarizedJob(job); |
| StringBuffer taskSummary = new StringBuffer(); |
| taskSummary.append("\nTask Summary"); |
| taskSummary.append("\n============================"); |
| taskSummary.append("\nKind\tTotal\t"); |
| taskSummary.append("Successful\tFailed\tKilled\tStartTime\tFinishTime"); |
| taskSummary.append("\n"); |
| taskSummary.append("\nSetup\t").append(ts.totalSetups); |
| taskSummary.append("\t").append(ts.numFinishedSetups); |
| taskSummary.append("\t\t").append(ts.numFailedSetups); |
| taskSummary.append("\t").append(ts.numKilledSetups); |
| taskSummary.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, ts.setupStarted, 0)); |
| taskSummary.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, ts.setupFinished, ts.setupStarted)); |
| taskSummary.append("\nMap\t").append(ts.totalMaps); |
| taskSummary.append("\t").append(job.getFinishedMaps()); |
| taskSummary.append("\t\t").append(ts.numFailedMaps); |
| taskSummary.append("\t").append(ts.numKilledMaps); |
| taskSummary.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, ts.mapStarted, 0)); |
| taskSummary.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, ts.mapFinished, ts.mapStarted)); |
| taskSummary.append("\nReduce\t").append(ts.totalReduces); |
| taskSummary.append("\t").append(job.getFinishedReduces()); |
| taskSummary.append("\t\t").append(ts.numFailedReduces); |
| taskSummary.append("\t").append(ts.numKilledReduces); |
| taskSummary.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, ts.reduceStarted, 0)); |
| taskSummary.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, ts.reduceFinished, ts.reduceStarted)); |
| taskSummary.append("\nCleanup\t").append(ts.totalCleanups); |
| taskSummary.append("\t").append(ts.numFinishedCleanups); |
| taskSummary.append("\t\t").append(ts.numFailedCleanups); |
| taskSummary.append("\t").append(ts.numKilledCleanups); |
| taskSummary.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, ts.cleanupStarted, 0)); |
| taskSummary.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, ts.cleanupFinished, |
| ts.cleanupStarted)); |
| taskSummary.append("\n============================\n"); |
| System.out.println(taskSummary.toString()); |
| } |
| |
| private void printJobAnalysis() { |
| if (!job.getJobStatus().equals |
| (JobStatus.getJobRunState(JobStatus.SUCCEEDED))) { |
| System.out.println("No Analysis available as job did not finish"); |
| return; |
| } |
| |
| AnalyzedJob avg = new AnalyzedJob(job); |
| |
| System.out.println("\nAnalysis"); |
| System.out.println("========="); |
| printAnalysis(avg.getMapTasks(), cMap, "map", avg.getAvgMapTime(), 10); |
| printLast(avg.getMapTasks(), "map", cFinishMapRed); |
| |
| if (avg.getReduceTasks().length > 0) { |
| printAnalysis(avg.getReduceTasks(), cShuffle, "shuffle", |
| avg.getAvgShuffleTime(), 10); |
| printLast(avg.getReduceTasks(), "shuffle", cFinishShuffle); |
| |
| printAnalysis(avg.getReduceTasks(), cReduce, "reduce", |
| avg.getAvgReduceTime(), 10); |
| printLast(avg.getReduceTasks(), "reduce", cFinishMapRed); |
| } |
| System.out.println("========="); |
| } |
| |
| private void printAnalysis(JobHistoryParser.TaskAttemptInfo [] tasks, |
| Comparator<JobHistoryParser.TaskAttemptInfo> cmp, |
| String taskType, |
| long avg, |
| int showTasks) { |
| Arrays.sort(tasks, cmp); |
| JobHistoryParser.TaskAttemptInfo min = tasks[tasks.length-1]; |
| StringBuffer details = new StringBuffer(); |
| details.append("\nTime taken by best performing "); |
| details.append(taskType).append(" task "); |
| details.append(min.getAttemptId().getTaskID().toString()).append(": "); |
| if ("map".equals(taskType)) { |
| details.append(StringUtils.formatTimeDiff( |
| min.getFinishTime(), |
| min.getStartTime())); |
| } else if ("shuffle".equals(taskType)) { |
| details.append(StringUtils.formatTimeDiff( |
| min.getShuffleFinishTime(), |
| min.getStartTime())); |
| } else { |
| details.append(StringUtils.formatTimeDiff( |
| min.getFinishTime(), |
| min.getShuffleFinishTime())); |
| } |
| details.append("\nAverage time taken by "); |
| details.append(taskType).append(" tasks: "); |
| details.append(StringUtils.formatTimeDiff(avg, 0)); |
| details.append("\nWorse performing "); |
| details.append(taskType).append(" tasks: "); |
| details.append("\nTaskId\t\tTimetaken"); |
| System.out.println(details.toString()); |
| for (int i = 0; i < showTasks && i < tasks.length; i++) { |
| details.setLength(0); |
| details.append(tasks[i].getAttemptId().getTaskID()).append(" "); |
| if ("map".equals(taskType)) { |
| details.append(StringUtils.formatTimeDiff( |
| tasks[i].getFinishTime(), |
| tasks[i].getStartTime())); |
| } else if ("shuffle".equals(taskType)) { |
| details.append(StringUtils.formatTimeDiff( |
| tasks[i].getShuffleFinishTime(), |
| tasks[i].getStartTime())); |
| } else { |
| details.append(StringUtils.formatTimeDiff( |
| tasks[i].getFinishTime(), |
| tasks[i].getShuffleFinishTime())); |
| } |
| System.out.println(details.toString()); |
| } |
| } |
| |
| private void printLast(JobHistoryParser.TaskAttemptInfo [] tasks, |
| String taskType, |
| Comparator<JobHistoryParser.TaskAttemptInfo> cmp |
| ) { |
| Arrays.sort(tasks, cFinishMapRed); |
| JobHistoryParser.TaskAttemptInfo last = tasks[0]; |
| StringBuffer lastBuf = new StringBuffer(); |
| lastBuf.append("The last ").append(taskType); |
| lastBuf.append(" task ").append(last.getAttemptId().getTaskID()); |
| Long finishTime; |
| if ("shuffle".equals(taskType)) { |
| finishTime = last.getShuffleFinishTime(); |
| } else { |
| finishTime = last.getFinishTime(); |
| } |
| lastBuf.append(" finished at (relative to the Job launch time): "); |
| lastBuf.append(StringUtils.getFormattedTimeWithDiff(dateFormat, |
| finishTime, job.getLaunchTime())); |
| System.out.println(lastBuf.toString()); |
| } |
| |
| private void printTasks(TaskType taskType, String status) { |
| Map<TaskID, JobHistoryParser.TaskInfo> tasks = job.getAllTasks(); |
| StringBuffer header = new StringBuffer(); |
| header.append("\n").append(status).append(" "); |
| header.append(taskType).append(" task list for ").append(jobId); |
| header.append("\nTaskId\t\tStartTime\tFinishTime\tError"); |
| if (TaskType.MAP.equals(taskType)) { |
| header.append("\tInputSplits"); |
| } |
| header.append("\n===================================================="); |
| StringBuffer taskList = new StringBuffer(); |
| for (JobHistoryParser.TaskInfo task : tasks.values()) { |
| if (taskType.equals(task.getTaskType()) && |
| (status.equals(task.getTaskStatus()) |
| || status.equalsIgnoreCase("ALL"))) { |
| taskList.setLength(0); |
| taskList.append(task.getTaskId()); |
| taskList.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, task.getStartTime(), 0)); |
| taskList.append("\t").append(StringUtils.getFormattedTimeWithDiff( |
| dateFormat, task.getFinishTime(), |
| task.getStartTime())); |
| taskList.append("\t").append(task.getError()); |
| if (TaskType.MAP.equals(taskType)) { |
| taskList.append("\t").append(task.getSplitLocations()); |
| } |
| if (taskList != null) { |
| System.out.println(header.toString()); |
| System.out.println(taskList.toString()); |
| } |
| } |
| } |
| } |
| |
| private void printFailedAttempts(FilteredJob filteredJob) { |
| Map<String, Set<TaskID>> badNodes = filteredJob.getFilteredMap(); |
| StringBuffer attempts = new StringBuffer(); |
| if (badNodes.size() > 0) { |
| attempts.append("\n").append(filteredJob.getFilter()); |
| attempts.append(" task attempts by nodes"); |
| attempts.append("\nHostname\tFailedTasks"); |
| attempts.append("\n==============================="); |
| System.out.println(attempts.toString()); |
| for (Map.Entry<String, |
| Set<TaskID>> entry : badNodes.entrySet()) { |
| String node = entry.getKey(); |
| Set<TaskID> failedTasks = entry.getValue(); |
| attempts.setLength(0); |
| attempts.append(node).append("\t"); |
| for (TaskID t : failedTasks) { |
| attempts.append(t).append(", "); |
| } |
| System.out.println(attempts.toString()); |
| } |
| } |
| } |
| |
| /** |
| * Return the TaskLogsUrl of a particular TaskAttempt |
| * |
| * @param attempt |
| * @return the taskLogsUrl. null if http-port or tracker-name or |
| * task-attempt-id are unavailable. |
| */ |
| public static String getTaskLogsUrl( |
| JobHistoryParser.TaskAttemptInfo attempt) { |
| if (attempt.getHttpPort() == -1 |
| || attempt.getTrackerName().equals("") |
| || attempt.getAttemptId() == null) { |
| return null; |
| } |
| |
| String taskTrackerName = |
| HostUtil.convertTrackerNameToHostName( |
| attempt.getTrackerName()); |
| return HostUtil.getTaskLogUrl(taskTrackerName, |
| Integer.toString(attempt.getHttpPort()), |
| attempt.getAttemptId().toString()); |
| } |
| |
| private Comparator<JobHistoryParser.TaskAttemptInfo> cMap = |
| new Comparator<JobHistoryParser.TaskAttemptInfo>() { |
| public int compare(JobHistoryParser.TaskAttemptInfo t1, |
| JobHistoryParser.TaskAttemptInfo t2) { |
| long l1 = t1.getFinishTime() - t1.getStartTime(); |
| long l2 = t2.getFinishTime() - t2.getStartTime(); |
| return (l2 < l1 ? -1 : (l2 == l1 ? 0 : 1)); |
| } |
| }; |
| |
| private Comparator<JobHistoryParser.TaskAttemptInfo> cShuffle = |
| new Comparator<JobHistoryParser.TaskAttemptInfo>() { |
| public int compare(JobHistoryParser.TaskAttemptInfo t1, |
| JobHistoryParser.TaskAttemptInfo t2) { |
| long l1 = t1.getShuffleFinishTime() - t1.getStartTime(); |
| long l2 = t2.getShuffleFinishTime() - t2.getStartTime(); |
| return (l2 < l1 ? -1 : (l2 == l1 ? 0 : 1)); |
| } |
| }; |
| |
| private Comparator<JobHistoryParser.TaskAttemptInfo> cFinishShuffle = |
| new Comparator<JobHistoryParser.TaskAttemptInfo>() { |
| public int compare(JobHistoryParser.TaskAttemptInfo t1, |
| JobHistoryParser.TaskAttemptInfo t2) { |
| long l1 = t1.getShuffleFinishTime(); |
| long l2 = t2.getShuffleFinishTime(); |
| return (l2 < l1 ? -1 : (l2 == l1 ? 0 : 1)); |
| } |
| }; |
| |
| private Comparator<JobHistoryParser.TaskAttemptInfo> cFinishMapRed = |
| new Comparator<JobHistoryParser.TaskAttemptInfo>() { |
| public int compare(JobHistoryParser.TaskAttemptInfo t1, |
| JobHistoryParser.TaskAttemptInfo t2) { |
| long l1 = t1.getFinishTime(); |
| long l2 = t2.getFinishTime(); |
| return (l2 < l1 ? -1 : (l2 == l1 ? 0 : 1)); |
| } |
| }; |
| |
| private Comparator<JobHistoryParser.TaskAttemptInfo> cReduce = |
| new Comparator<JobHistoryParser.TaskAttemptInfo>() { |
| public int compare(JobHistoryParser.TaskAttemptInfo t1, |
| JobHistoryParser.TaskAttemptInfo t2) { |
| long l1 = t1.getFinishTime() - |
| t1.getShuffleFinishTime(); |
| long l2 = t2.getFinishTime() - |
| t2.getShuffleFinishTime(); |
| return (l2 < l1 ? -1 : (l2 == l1 ? 0 : 1)); |
| } |
| }; |
| |
| /** |
| * Utility class used the summarize the job. |
| * Used by HistoryViewer and the JobHistory UI. |
| * |
| */ |
| public static class SummarizedJob { |
| Map<TaskID, JobHistoryParser.TaskInfo> tasks; |
| int totalMaps = 0; |
| int totalReduces = 0; |
| int totalCleanups = 0; |
| int totalSetups = 0; |
| int numFailedMaps = 0; |
| int numKilledMaps = 0; |
| int numFailedReduces = 0; |
| int numKilledReduces = 0; |
| int numFinishedCleanups = 0; |
| int numFailedCleanups = 0; |
| int numKilledCleanups = 0; |
| int numFinishedSetups = 0; |
| int numFailedSetups = 0; |
| int numKilledSetups = 0; |
| long mapStarted = 0; |
| long mapFinished = 0; |
| long reduceStarted = 0; |
| long reduceFinished = 0; |
| long cleanupStarted = 0; |
| long cleanupFinished = 0; |
| long setupStarted = 0; |
| long setupFinished = 0; |
| |
| /** Get total maps */ |
| public int getTotalMaps() { return totalMaps; } |
| /** Get total reduces */ |
| public int getTotalReduces() { return totalReduces; } |
| /** Get number of clean up tasks */ |
| public int getTotalCleanups() { return totalCleanups; } |
| /** Get number of set up tasks */ |
| public int getTotalSetups() { return totalSetups; } |
| /** Get number of failed maps */ |
| public int getNumFailedMaps() { return numFailedMaps; } |
| /** Get number of killed maps */ |
| public int getNumKilledMaps() { return numKilledMaps; } |
| /** Get number of failed reduces */ |
| public int getNumFailedReduces() { return numFailedReduces; } |
| /** Get number of killed reduces */ |
| public int getNumKilledReduces() { return numKilledReduces; } |
| /** Get number of cleanup tasks that finished */ |
| public int getNumFinishedCleanups() { return numFinishedCleanups; } |
| /** Get number of failed cleanup tasks */ |
| public int getNumFailedCleanups() { return numFailedCleanups; } |
| /** Get number of killed cleanup tasks */ |
| public int getNumKilledCleanups() { return numKilledCleanups; } |
| /** Get number of finished set up tasks */ |
| public int getNumFinishedSetups() { return numFinishedSetups; } |
| /** Get number of failed set up tasks */ |
| public int getNumFailedSetups() { return numFailedSetups; } |
| /** Get number of killed set up tasks */ |
| public int getNumKilledSetups() { return numKilledSetups; } |
| /** Get number of maps that were started */ |
| public long getMapStarted() { return mapStarted; } |
| /** Get number of maps that finished */ |
| public long getMapFinished() { return mapFinished; } |
| /** Get number of Reducers that were started */ |
| public long getReduceStarted() { return reduceStarted; } |
| /** Get number of reducers that finished */ |
| public long getReduceFinished() { return reduceFinished; } |
| /** Get number of cleanup tasks started */ |
| public long getCleanupStarted() { return cleanupStarted; } |
| /** Get number of cleanup tasks that finished */ |
| public long getCleanupFinished() { return cleanupFinished; } |
| /** Get number of setup tasks that started */ |
| public long getSetupStarted() { return setupStarted; } |
| /** Get number of setup tasks that finished */ |
| public long getSetupFinished() { return setupFinished; } |
| |
| /** Create summary information for the parsed job */ |
| public SummarizedJob(JobInfo job) { |
| tasks = job.getAllTasks(); |
| |
| for (JobHistoryParser.TaskInfo task : tasks.values()) { |
| Map<TaskAttemptID, JobHistoryParser.TaskAttemptInfo> attempts = |
| task.getAllTaskAttempts(); |
| //allHosts.put(task.getHo(Keys.HOSTNAME), ""); |
| for (JobHistoryParser.TaskAttemptInfo attempt : attempts.values()) { |
| long startTime = attempt.getStartTime(); |
| long finishTime = attempt.getFinishTime(); |
| if (attempt.getTaskType().equals(TaskType.MAP)) { |
| if (mapStarted== 0 || mapStarted > startTime) { |
| mapStarted = startTime; |
| } |
| if (mapFinished < finishTime) { |
| mapFinished = finishTime; |
| } |
| totalMaps++; |
| if (attempt.getTaskStatus().equals |
| (TaskStatus.State.FAILED.toString())) { |
| numFailedMaps++; |
| } else if (attempt.getTaskStatus().equals |
| (TaskStatus.State.KILLED.toString())) { |
| numKilledMaps++; |
| } |
| } else if (attempt.getTaskType().equals(TaskType.REDUCE)) { |
| if (reduceStarted==0||reduceStarted > startTime) { |
| reduceStarted = startTime; |
| } |
| if (reduceFinished < finishTime) { |
| reduceFinished = finishTime; |
| } |
| totalReduces++; |
| if (attempt.getTaskStatus().equals |
| (TaskStatus.State.FAILED.toString())) { |
| numFailedReduces++; |
| } else if (attempt.getTaskStatus().equals |
| (TaskStatus.State.KILLED.toString())) { |
| numKilledReduces++; |
| } |
| } else if (attempt.getTaskType().equals(TaskType.JOB_CLEANUP)) { |
| if (cleanupStarted==0||cleanupStarted > startTime) { |
| cleanupStarted = startTime; |
| } |
| if (cleanupFinished < finishTime) { |
| cleanupFinished = finishTime; |
| } |
| totalCleanups++; |
| if (attempt.getTaskStatus().equals |
| (TaskStatus.State.SUCCEEDED.toString())) { |
| numFinishedCleanups++; |
| } else if (attempt.getTaskStatus().equals |
| (TaskStatus.State.FAILED.toString())) { |
| numFailedCleanups++; |
| } else if (attempt.getTaskStatus().equals |
| (TaskStatus.State.KILLED.toString())) { |
| numKilledCleanups++; |
| } |
| } else if (attempt.getTaskType().equals(TaskType.JOB_SETUP)) { |
| if (setupStarted==0||setupStarted > startTime) { |
| setupStarted = startTime; |
| } |
| if (setupFinished < finishTime) { |
| setupFinished = finishTime; |
| } |
| totalSetups++; |
| if (attempt.getTaskStatus().equals |
| (TaskStatus.State.SUCCEEDED.toString())) { |
| numFinishedSetups++; |
| } else if (attempt.getTaskStatus().equals |
| (TaskStatus.State.FAILED.toString())) { |
| numFailedSetups++; |
| } else if (attempt.getTaskStatus().equals |
| (TaskStatus.State.KILLED.toString())) { |
| numKilledSetups++; |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| /** |
| * Utility class used while analyzing the job. |
| * Used by HistoryViewer and the JobHistory UI. |
| */ |
| |
| public static class AnalyzedJob { |
| private long avgMapTime; |
| private long avgReduceTime; |
| private long avgShuffleTime; |
| |
| private JobHistoryParser.TaskAttemptInfo [] mapTasks; |
| private JobHistoryParser.TaskAttemptInfo [] reduceTasks; |
| |
| /** Get the average map time */ |
| public long getAvgMapTime() { return avgMapTime; } |
| /** Get the average reduce time */ |
| public long getAvgReduceTime() { return avgReduceTime; } |
| /** Get the average shuffle time */ |
| public long getAvgShuffleTime() { return avgShuffleTime; } |
| /** Get the map tasks list */ |
| public JobHistoryParser.TaskAttemptInfo [] getMapTasks() { |
| return mapTasks; |
| } |
| /** Get the reduce tasks list */ |
| public JobHistoryParser.TaskAttemptInfo [] getReduceTasks() { |
| return reduceTasks; |
| } |
| /** Generate analysis information for the parsed job */ |
| public AnalyzedJob (JobInfo job) { |
| Map<TaskID, JobHistoryParser.TaskInfo> tasks = job.getAllTasks(); |
| int finishedMaps = (int) job.getFinishedMaps(); |
| int finishedReduces = (int) job.getFinishedReduces(); |
| mapTasks = |
| new JobHistoryParser.TaskAttemptInfo[finishedMaps]; |
| reduceTasks = |
| new JobHistoryParser.TaskAttemptInfo[finishedReduces]; |
| int mapIndex = 0 , reduceIndex=0; |
| avgMapTime = 0; |
| avgReduceTime = 0; |
| avgShuffleTime = 0; |
| |
| for (JobHistoryParser.TaskInfo task : tasks.values()) { |
| Map<TaskAttemptID, JobHistoryParser.TaskAttemptInfo> attempts = |
| task.getAllTaskAttempts(); |
| for (JobHistoryParser.TaskAttemptInfo attempt : attempts.values()) { |
| if (attempt.getTaskStatus(). |
| equals(TaskStatus.State.SUCCEEDED.toString())) { |
| long avgFinishTime = (attempt.getFinishTime() - |
| attempt.getStartTime()); |
| if (attempt.getTaskType().equals(TaskType.MAP)) { |
| mapTasks[mapIndex++] = attempt; |
| avgMapTime += avgFinishTime; |
| } else if (attempt.getTaskType().equals(TaskType.REDUCE)) { |
| reduceTasks[reduceIndex++] = attempt; |
| avgShuffleTime += (attempt.getShuffleFinishTime() - |
| attempt.getStartTime()); |
| avgReduceTime += (attempt.getFinishTime() - |
| attempt.getShuffleFinishTime()); |
| } |
| break; |
| } |
| } |
| } |
| if (finishedMaps > 0) { |
| avgMapTime /= finishedMaps; |
| } |
| if (finishedReduces > 0) { |
| avgReduceTime /= finishedReduces; |
| avgShuffleTime /= finishedReduces; |
| } |
| } |
| } |
| |
| /** |
| * Utility to filter out events based on the task status |
| * |
| */ |
| public static class FilteredJob { |
| |
| private Map<String, Set<TaskID>> badNodesToFilteredTasks = |
| new HashMap<String, Set<TaskID>>(); |
| |
| private String filter; |
| |
| /** Get the map of the filtered tasks */ |
| public Map<String, Set<TaskID>> getFilteredMap() { |
| return badNodesToFilteredTasks; |
| } |
| |
| /** Get the current filter */ |
| public String getFilter() { return filter; } |
| |
| /** Apply the filter (status) on the parsed job and generate summary */ |
| public FilteredJob(JobInfo job, String status) { |
| |
| filter = status; |
| |
| Map<TaskID, JobHistoryParser.TaskInfo> tasks = job.getAllTasks(); |
| |
| for (JobHistoryParser.TaskInfo task : tasks.values()) { |
| Map<TaskAttemptID, JobHistoryParser.TaskAttemptInfo> attempts = |
| task.getAllTaskAttempts(); |
| for (JobHistoryParser.TaskAttemptInfo attempt : attempts.values()) { |
| if (attempt.getTaskStatus().equals(status)) { |
| String hostname = attempt.getHostname(); |
| TaskID id = attempt.getAttemptId().getTaskID(); |
| |
| Set<TaskID> set = badNodesToFilteredTasks.get(hostname); |
| |
| if (set == null) { |
| set = new TreeSet<TaskID>(); |
| set.add(id); |
| badNodesToFilteredTasks.put(hostname, set); |
| }else{ |
| set.add(id); |
| } |
| } |
| } |
| } |
| } |
| } |
| } |
