| /** |
| * 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.fs; |
| |
| import java.io.BufferedReader; |
| import java.io.BufferedWriter; |
| import java.io.DataInputStream; |
| import java.io.DataOutputStream; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.InputStreamReader; |
| import java.io.OutputStreamWriter; |
| import java.text.SimpleDateFormat; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Date; |
| import java.util.Map; |
| import java.util.StringTokenizer; |
| import java.util.HashMap; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.io.LongWritable; |
| import org.apache.hadoop.io.SequenceFile; |
| import org.apache.hadoop.io.Text; |
| import org.apache.hadoop.io.SequenceFile.CompressionType; |
| import org.apache.hadoop.io.compress.CompressionCodec; |
| import org.apache.hadoop.io.compress.GzipCodec; |
| import org.apache.hadoop.mapred.*; |
| import org.apache.hadoop.util.ReflectionUtils; |
| import org.apache.hadoop.util.StringUtils; |
| |
| /** |
| * Job History Log Analyzer. |
| * |
| * <h3>Description.</h3> |
| * This a tool for parsing and analyzing history logs of map-reduce jobs. |
| * History logs contain information about execution of jobs, tasks, and |
| * attempts. This tool focuses on submission, launch, start, and finish times, |
| * as well as the success or failure of jobs, tasks, and attempts. |
| * <p> |
| * The analyzer calculates <em>per hour slot utilization</em> for the cluster |
| * as follows. |
| * For each task attempt it divides the time segment from the start of the |
| * attempt t<sub>S</sub> to the finish t<sub>F</sub> into whole hours |
| * [t<sub>0</sub>, ..., t<sub>n</sub>], where t<sub>0</sub> <= t<sub>S</sub> |
| * is the maximal whole hour preceding t<sub>S</sub>, and |
| * t<sub>n</sub> >= t<sub>F</sub> is the minimal whole hour after t<sub>F</sub>. |
| * Thus, [t<sub>0</sub>, ..., t<sub>n</sub>] covers the segment |
| * [t<sub>S</sub>, t<sub>F</sub>], during which the attempt was executed. |
| * Each interval [t<sub>i</sub>, t<sub>i+1</sub>] fully contained in |
| * [t<sub>S</sub>, t<sub>F</sub>] corresponds to exactly one slot on |
| * a map-reduce cluster (usually MAP-slot or REDUCE-slot). |
| * If interval [t<sub>i</sub>, t<sub>i+1</sub>] only intersects with |
| * [t<sub>S</sub>, t<sub>F</sub>] then we say that the task |
| * attempt used just a fraction of the slot during this hour. |
| * The fraction equals the size of the intersection. |
| * Let slotTime(A, h) denote the number of slots calculated that way for a |
| * specific attempt A during hour h. |
| * The tool then sums all slots for all attempts for every hour. |
| * The result is the slot hour utilization of the cluster: |
| * <tt>slotTime(h) = SUM<sub>A</sub> slotTime(A,h)</tt>. |
| * <p> |
| * Log analyzer calculates slot hours for <em>MAP</em> and <em>REDUCE</em> |
| * attempts separately. |
| * <p> |
| * Log analyzer distinguishes between <em>successful</em> and <em>failed</em> |
| * attempts. Task attempt is considered successful if its own status is SUCCESS |
| * and the statuses of the task and the job it is a part of are also SUCCESS. |
| * Otherwise the task attempt is considered failed. |
| * <p> |
| * Map-reduce clusters are usually configured to have a fixed number of MAP |
| * and REDUCE slots per node. Thus the maximal possible number of slots on |
| * the cluster is <tt>total_slots = total_nodes * slots_per_node</tt>. |
| * Effective slot hour cannot exceed <tt>total_slots</tt> for successful |
| * attempts. |
| * <p> |
| * <em>Pending time</em> characterizes the wait time of attempts. |
| * It is calculated similarly to the slot hour except that the wait interval |
| * starts when the job is submitted and ends when an attempt starts execution. |
| * In addition to that pending time also includes intervals between attempts |
| * of the same task if it was re-executed. |
| * <p> |
| * History log analyzer calculates two pending time variations. First is based |
| * on job submission time as described above, second, starts the wait interval |
| * when the job is launched rather than submitted. |
| * |
| * <h3>Input.</h3> |
| * The following input parameters can be specified in the argument string |
| * to the job log analyzer: |
| * <ul> |
| * <li><tt>-historyDir inputDir</tt> specifies the location of the directory |
| * where analyzer will be looking for job history log files.</li> |
| * <li><tt>-resFile resultFile</tt> the name of the result file.</li> |
| * <li><tt>-usersIncluded | -usersExcluded userList</tt> slot utilization and |
| * pending time can be calculated for all or for all but the specified users. |
| * <br> |
| * <tt>userList</tt> is a comma or semicolon separated list of users.</li> |
| * <li><tt>-gzip</tt> is used if history log files are compressed. |
| * Only {@link GzipCodec} is currently supported.</li> |
| * <li><tt>-jobDelimiter pattern</tt> one can concatenate original log files into |
| * larger file(s) with the specified delimiter to recognize the end of the log |
| * for one job from the next one.<br> |
| * <tt>pattern</tt> is a java regular expression |
| * {@link java.util.regex.Pattern}, which should match only the log delimiters. |
| * <br> |
| * E.g. pattern <tt>".!!FILE=.*!!"</tt> matches delimiters, which contain |
| * the original history log file names in the following form:<br> |
| * <tt>"$!!FILE=my.job.tracker.com_myJobId_user_wordcount.log!!"</tt></li> |
| * <li><tt>-clean</tt> cleans up default directories used by the analyzer.</li> |
| * <li><tt>-test</tt> test one file locally and exit; |
| * does not require map-reduce.</li> |
| * <li><tt>-help</tt> print usage.</li> |
| * </ul> |
| * |
| * <h3>Output.</h3> |
| * The output file is formatted as a tab separated table consisting of four |
| * columns: <tt>SERIES, PERIOD, TYPE, SLOT_HOUR</tt>. |
| * <ul> |
| * <li><tt>SERIES</tt> one of the four statistical series;</li> |
| * <li><tt>PERIOD</tt> the start of the time interval in the following format: |
| * <tt>"yyyy-mm-dd hh:mm:ss"</tt>;</li> |
| * <li><tt>TYPE</tt> the slot type, e.g. MAP or REDUCE;</li> |
| * <li><tt>SLOT_HOUR</tt> the value of the slot usage during this |
| * time interval.</li> |
| * </ul> |
| */ |
| @SuppressWarnings("deprecation") |
| public class JHLogAnalyzer { |
| private static final Log LOG = LogFactory.getLog(JHLogAnalyzer.class); |
| // Constants |
| private static final String JHLA_ROOT_DIR = |
| System.getProperty("test.build.data", "stats/JHLA"); |
| private static final Path INPUT_DIR = new Path(JHLA_ROOT_DIR, "jhla_input"); |
| private static final String BASE_INPUT_FILE_NAME = "jhla_in_"; |
| private static final Path OUTPUT_DIR = new Path(JHLA_ROOT_DIR, "jhla_output"); |
| private static final Path RESULT_FILE = |
| new Path(JHLA_ROOT_DIR, "jhla_result.txt"); |
| private static final Path DEFAULT_HISTORY_DIR = new Path("history"); |
| |
| private static final int DEFAULT_TIME_INTERVAL_MSEC = 1000*60*60; // 1 hour |
| |
| static{ |
| Configuration.addDefaultResource("hdfs-default.xml"); |
| Configuration.addDefaultResource("hdfs-site.xml"); |
| } |
| |
| static enum StatSeries { |
| STAT_ALL_SLOT_TIME |
| (AccumulatingReducer.VALUE_TYPE_LONG + "allSlotTime"), |
| STAT_FAILED_SLOT_TIME |
| (AccumulatingReducer.VALUE_TYPE_LONG + "failedSlotTime"), |
| STAT_SUBMIT_PENDING_SLOT_TIME |
| (AccumulatingReducer.VALUE_TYPE_LONG + "submitPendingSlotTime"), |
| STAT_LAUNCHED_PENDING_SLOT_TIME |
| (AccumulatingReducer.VALUE_TYPE_LONG + "launchedPendingSlotTime"); |
| |
| private String statName = null; |
| private StatSeries(String name) {this.statName = name;} |
| public String toString() {return statName;} |
| } |
| |
| private static class FileCreateDaemon extends Thread { |
| private static final int NUM_CREATE_THREADS = 10; |
| private static volatile int numFinishedThreads; |
| private static volatile int numRunningThreads; |
| private static FileStatus[] jhLogFiles; |
| |
| FileSystem fs; |
| int start; |
| int end; |
| |
| FileCreateDaemon(FileSystem fs, int start, int end) { |
| this.fs = fs; |
| this.start = start; |
| this.end = end; |
| } |
| |
| public void run() { |
| try { |
| for(int i=start; i < end; i++) { |
| String name = getFileName(i); |
| Path controlFile = new Path(INPUT_DIR, "in_file_" + name); |
| SequenceFile.Writer writer = null; |
| try { |
| writer = SequenceFile.createWriter(fs, fs.getConf(), controlFile, |
| Text.class, LongWritable.class, |
| CompressionType.NONE); |
| String logFile = jhLogFiles[i].getPath().toString(); |
| writer.append(new Text(logFile), new LongWritable(0)); |
| } catch(Exception e) { |
| throw new IOException(e); |
| } finally { |
| if (writer != null) |
| writer.close(); |
| writer = null; |
| } |
| } |
| } catch(IOException ex) { |
| LOG.error("FileCreateDaemon failed.", ex); |
| } |
| numFinishedThreads++; |
| } |
| |
| private static void createControlFile(FileSystem fs, Path jhLogDir |
| ) throws IOException { |
| fs.delete(INPUT_DIR, true); |
| jhLogFiles = fs.listStatus(jhLogDir); |
| |
| numFinishedThreads = 0; |
| try { |
| int start = 0; |
| int step = jhLogFiles.length / NUM_CREATE_THREADS |
| + ((jhLogFiles.length % NUM_CREATE_THREADS) > 0 ? 1 : 0); |
| FileCreateDaemon[] daemons = new FileCreateDaemon[NUM_CREATE_THREADS]; |
| numRunningThreads = 0; |
| for(int tIdx=0; tIdx < NUM_CREATE_THREADS && start < jhLogFiles.length; tIdx++) { |
| int end = Math.min(start + step, jhLogFiles.length); |
| daemons[tIdx] = new FileCreateDaemon(fs, start, end); |
| start += step; |
| numRunningThreads++; |
| } |
| for(int tIdx=0; tIdx < numRunningThreads; tIdx++) { |
| daemons[tIdx].start(); |
| } |
| } finally { |
| int prevValue = 0; |
| while(numFinishedThreads < numRunningThreads) { |
| if(prevValue < numFinishedThreads) { |
| LOG.info("Finished " + numFinishedThreads + " threads out of " + numRunningThreads); |
| prevValue = numFinishedThreads; |
| } |
| try {Thread.sleep(500);} catch (InterruptedException e) {} |
| } |
| } |
| } |
| } |
| |
| private static void createControlFile(FileSystem fs, Path jhLogDir |
| ) throws IOException { |
| LOG.info("creating control file: JH log dir = " + jhLogDir); |
| FileCreateDaemon.createControlFile(fs, jhLogDir); |
| LOG.info("created control file: JH log dir = " + jhLogDir); |
| } |
| |
| private static String getFileName(int fIdx) { |
| return BASE_INPUT_FILE_NAME + Integer.toString(fIdx); |
| } |
| |
| /** |
| * If keyVal is of the form KEY="VALUE", then this will return [KEY, VALUE] |
| */ |
| private static String [] getKeyValue(String t) throws IOException { |
| String[] keyVal = t.split("=\"*|\""); |
| return keyVal; |
| } |
| |
| /** |
| * JobHistory log record. |
| */ |
| private static class JobHistoryLog { |
| String JOBID; |
| String JOB_STATUS; |
| long SUBMIT_TIME; |
| long LAUNCH_TIME; |
| long FINISH_TIME; |
| long TOTAL_MAPS; |
| long TOTAL_REDUCES; |
| long FINISHED_MAPS; |
| long FINISHED_REDUCES; |
| String USER; |
| Map<String, TaskHistoryLog> tasks; |
| |
| boolean isSuccessful() { |
| return (JOB_STATUS != null) && JOB_STATUS.equals("SUCCESS"); |
| } |
| |
| void parseLine(String line) throws IOException { |
| StringTokenizer tokens = new StringTokenizer(line); |
| if(!tokens.hasMoreTokens()) |
| return; |
| String what = tokens.nextToken(); |
| // Line should start with one of the following: |
| // Job, Task, MapAttempt, ReduceAttempt |
| if(what.equals("Job")) |
| updateJob(tokens); |
| else if(what.equals("Task")) |
| updateTask(tokens); |
| else if(what.indexOf("Attempt") >= 0) |
| updateTaskAttempt(tokens); |
| } |
| |
| private void updateJob(StringTokenizer tokens) throws IOException { |
| while(tokens.hasMoreTokens()) { |
| String t = tokens.nextToken(); |
| String[] keyVal = getKeyValue(t); |
| if(keyVal.length < 2) continue; |
| |
| if(keyVal[0].equals("JOBID")) { |
| if(JOBID == null) |
| JOBID = new String(keyVal[1]); |
| else if(!JOBID.equals(keyVal[1])) { |
| LOG.error("Incorrect JOBID: " |
| + keyVal[1].substring(0, Math.min(keyVal[1].length(), 100)) |
| + " expect " + JOBID); |
| return; |
| } |
| } |
| else if(keyVal[0].equals("JOB_STATUS")) |
| JOB_STATUS = new String(keyVal[1]); |
| else if(keyVal[0].equals("SUBMIT_TIME")) |
| SUBMIT_TIME = Long.parseLong(keyVal[1]); |
| else if(keyVal[0].equals("LAUNCH_TIME")) |
| LAUNCH_TIME = Long.parseLong(keyVal[1]); |
| else if(keyVal[0].equals("FINISH_TIME")) |
| FINISH_TIME = Long.parseLong(keyVal[1]); |
| else if(keyVal[0].equals("TOTAL_MAPS")) |
| TOTAL_MAPS = Long.parseLong(keyVal[1]); |
| else if(keyVal[0].equals("TOTAL_REDUCES")) |
| TOTAL_REDUCES = Long.parseLong(keyVal[1]); |
| else if(keyVal[0].equals("FINISHED_MAPS")) |
| FINISHED_MAPS = Long.parseLong(keyVal[1]); |
| else if(keyVal[0].equals("FINISHED_REDUCES")) |
| FINISHED_REDUCES = Long.parseLong(keyVal[1]); |
| else if(keyVal[0].equals("USER")) |
| USER = new String(keyVal[1]); |
| } |
| } |
| |
| private void updateTask(StringTokenizer tokens) throws IOException { |
| // unpack |
| TaskHistoryLog task = new TaskHistoryLog().parse(tokens); |
| if(task.TASKID == null) { |
| LOG.error("TASKID = NULL for job " + JOBID); |
| return; |
| } |
| // update or insert |
| if(tasks == null) |
| tasks = new HashMap<String, TaskHistoryLog>((int)(TOTAL_MAPS + TOTAL_REDUCES)); |
| TaskHistoryLog existing = tasks.get(task.TASKID); |
| if(existing == null) |
| tasks.put(task.TASKID, task); |
| else |
| existing.updateWith(task); |
| } |
| |
| private void updateTaskAttempt(StringTokenizer tokens) throws IOException { |
| // unpack |
| TaskAttemptHistoryLog attempt = new TaskAttemptHistoryLog(); |
| String taskID = attempt.parse(tokens); |
| if(taskID == null) return; |
| if(tasks == null) |
| tasks = new HashMap<String, TaskHistoryLog>((int)(TOTAL_MAPS + TOTAL_REDUCES)); |
| TaskHistoryLog existing = tasks.get(taskID); |
| if(existing == null) { |
| existing = new TaskHistoryLog(taskID); |
| tasks.put(taskID, existing); |
| } |
| existing.updateWith(attempt); |
| } |
| } |
| |
| /** |
| * TaskHistory log record. |
| */ |
| private static class TaskHistoryLog { |
| String TASKID; |
| String TASK_TYPE; // MAP, REDUCE, SETUP, CLEANUP |
| String TASK_STATUS; |
| long START_TIME; |
| long FINISH_TIME; |
| Map<String, TaskAttemptHistoryLog> attempts; |
| |
| TaskHistoryLog() {} |
| |
| TaskHistoryLog(String taskID) { |
| TASKID = taskID; |
| } |
| |
| boolean isSuccessful() { |
| return (TASK_STATUS != null) && TASK_STATUS.equals("SUCCESS"); |
| } |
| |
| TaskHistoryLog parse(StringTokenizer tokens) throws IOException { |
| while(tokens.hasMoreTokens()) { |
| String t = tokens.nextToken(); |
| String[] keyVal = getKeyValue(t); |
| if(keyVal.length < 2) continue; |
| |
| if(keyVal[0].equals("TASKID")) { |
| if(TASKID == null) |
| TASKID = new String(keyVal[1]); |
| else if(!TASKID.equals(keyVal[1])) { |
| LOG.error("Incorrect TASKID: " |
| + keyVal[1].substring(0, Math.min(keyVal[1].length(), 100)) |
| + " expect " + TASKID); |
| continue; |
| } |
| } |
| else if(keyVal[0].equals("TASK_TYPE")) |
| TASK_TYPE = new String(keyVal[1]); |
| else if(keyVal[0].equals("TASK_STATUS")) |
| TASK_STATUS = new String(keyVal[1]); |
| else if(keyVal[0].equals("START_TIME")) |
| START_TIME = Long.parseLong(keyVal[1]); |
| else if(keyVal[0].equals("FINISH_TIME")) |
| FINISH_TIME = Long.parseLong(keyVal[1]); |
| } |
| return this; |
| } |
| |
| /** |
| * Update with non-null fields of the same task log record. |
| */ |
| void updateWith(TaskHistoryLog from) throws IOException { |
| if(TASKID == null) |
| TASKID = from.TASKID; |
| else if(!TASKID.equals(from.TASKID)) { |
| throw new IOException("Incorrect TASKID: " + from.TASKID |
| + " expect " + TASKID); |
| } |
| if(TASK_TYPE == null) |
| TASK_TYPE = from.TASK_TYPE; |
| else if(! TASK_TYPE.equals(from.TASK_TYPE)) { |
| LOG.error( |
| "Incorrect TASK_TYPE: " + from.TASK_TYPE + " expect " + TASK_TYPE |
| + " for task " + TASKID); |
| return; |
| } |
| if(from.TASK_STATUS != null) |
| TASK_STATUS = from.TASK_STATUS; |
| if(from.START_TIME > 0) |
| START_TIME = from.START_TIME; |
| if(from.FINISH_TIME > 0) |
| FINISH_TIME = from.FINISH_TIME; |
| } |
| |
| /** |
| * Update with non-null fields of the task attempt log record. |
| */ |
| void updateWith(TaskAttemptHistoryLog attempt) throws IOException { |
| if(attempt.TASK_ATTEMPT_ID == null) { |
| LOG.error("Unexpected TASK_ATTEMPT_ID = null for task " + TASKID); |
| return; |
| } |
| if(attempts == null) |
| attempts = new HashMap<String, TaskAttemptHistoryLog>(); |
| TaskAttemptHistoryLog existing = attempts.get(attempt.TASK_ATTEMPT_ID); |
| if(existing == null) |
| attempts.put(attempt.TASK_ATTEMPT_ID, attempt); |
| else |
| existing.updateWith(attempt); |
| // update task start time |
| if(attempt.START_TIME > 0 && |
| (this.START_TIME == 0 || this.START_TIME > attempt.START_TIME)) |
| START_TIME = attempt.START_TIME; |
| } |
| } |
| |
| /** |
| * TaskAttemptHistory log record. |
| */ |
| private static class TaskAttemptHistoryLog { |
| String TASK_ATTEMPT_ID; |
| String TASK_STATUS; // this task attempt status |
| long START_TIME; |
| long FINISH_TIME; |
| long HDFS_BYTES_READ; |
| long HDFS_BYTES_WRITTEN; |
| long FILE_BYTES_READ; |
| long FILE_BYTES_WRITTEN; |
| |
| /** |
| * Task attempt is considered successful iff all three statuses |
| * of the attempt, the task, and the job equal "SUCCESS". |
| */ |
| boolean isSuccessful() { |
| return (TASK_STATUS != null) && TASK_STATUS.equals("SUCCESS"); |
| } |
| |
| String parse(StringTokenizer tokens) throws IOException { |
| String taskID = null; |
| while(tokens.hasMoreTokens()) { |
| String t = tokens.nextToken(); |
| String[] keyVal = getKeyValue(t); |
| if(keyVal.length < 2) continue; |
| |
| if(keyVal[0].equals("TASKID")) { |
| if(taskID == null) |
| taskID = new String(keyVal[1]); |
| else if(!taskID.equals(keyVal[1])) { |
| LOG.error("Incorrect TASKID: " + keyVal[1] + " expect " + taskID); |
| continue; |
| } |
| } |
| else if(keyVal[0].equals("TASK_ATTEMPT_ID")) { |
| if(TASK_ATTEMPT_ID == null) |
| TASK_ATTEMPT_ID = new String(keyVal[1]); |
| else if(!TASK_ATTEMPT_ID.equals(keyVal[1])) { |
| LOG.error("Incorrect TASKID: " + keyVal[1] + " expect " + taskID); |
| continue; |
| } |
| } |
| else if(keyVal[0].equals("TASK_STATUS")) |
| TASK_STATUS = new String(keyVal[1]); |
| else if(keyVal[0].equals("START_TIME")) |
| START_TIME = Long.parseLong(keyVal[1]); |
| else if(keyVal[0].equals("FINISH_TIME")) |
| FINISH_TIME = Long.parseLong(keyVal[1]); |
| } |
| return taskID; |
| } |
| |
| /** |
| * Update with non-null fields of the same task attempt log record. |
| */ |
| void updateWith(TaskAttemptHistoryLog from) throws IOException { |
| if(TASK_ATTEMPT_ID == null) |
| TASK_ATTEMPT_ID = from.TASK_ATTEMPT_ID; |
| else if(! TASK_ATTEMPT_ID.equals(from.TASK_ATTEMPT_ID)) { |
| throw new IOException( |
| "Incorrect TASK_ATTEMPT_ID: " + from.TASK_ATTEMPT_ID |
| + " expect " + TASK_ATTEMPT_ID); |
| } |
| if(from.TASK_STATUS != null) |
| TASK_STATUS = from.TASK_STATUS; |
| if(from.START_TIME > 0) |
| START_TIME = from.START_TIME; |
| if(from.FINISH_TIME > 0) |
| FINISH_TIME = from.FINISH_TIME; |
| if(from.HDFS_BYTES_READ > 0) |
| HDFS_BYTES_READ = from.HDFS_BYTES_READ; |
| if(from.HDFS_BYTES_WRITTEN > 0) |
| HDFS_BYTES_WRITTEN = from.HDFS_BYTES_WRITTEN; |
| if(from.FILE_BYTES_READ > 0) |
| FILE_BYTES_READ = from.FILE_BYTES_READ; |
| if(from.FILE_BYTES_WRITTEN > 0) |
| FILE_BYTES_WRITTEN = from.FILE_BYTES_WRITTEN; |
| } |
| } |
| |
| /** |
| * Key = statName*date-time*taskType |
| * Value = number of msec for the our |
| */ |
| private static class IntervalKey { |
| static final String KEY_FIELD_DELIMITER = "*"; |
| String statName; |
| String dateTime; |
| String taskType; |
| |
| IntervalKey(String stat, long timeMSec, String taskType) { |
| statName = stat; |
| SimpleDateFormat dateF = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); |
| dateTime = dateF.format(new Date(timeMSec)); |
| this.taskType = taskType; |
| } |
| |
| IntervalKey(String key) { |
| StringTokenizer keyTokens = new StringTokenizer(key, KEY_FIELD_DELIMITER); |
| if(!keyTokens.hasMoreTokens()) return; |
| statName = keyTokens.nextToken(); |
| if(!keyTokens.hasMoreTokens()) return; |
| dateTime = keyTokens.nextToken(); |
| if(!keyTokens.hasMoreTokens()) return; |
| taskType = keyTokens.nextToken(); |
| } |
| |
| void setStatName(String stat) { |
| statName = stat; |
| } |
| |
| String getStringKey() { |
| return statName + KEY_FIELD_DELIMITER + |
| dateTime + KEY_FIELD_DELIMITER + |
| taskType; |
| } |
| |
| Text getTextKey() { |
| return new Text(getStringKey()); |
| } |
| |
| public String toString() { |
| return getStringKey(); |
| } |
| } |
| |
| /** |
| * Mapper class. |
| */ |
| private static class JHLAMapper extends IOMapperBase<Object> { |
| /** |
| * A line pattern, which delimits history logs of different jobs, |
| * if multiple job logs are written in the same file. |
| * Null value means only one job log per file is expected. |
| * The pattern should be a regular expression as in |
| * {@link String#matches(String)}. |
| */ |
| String jobDelimiterPattern; |
| int maxJobDelimiterLineLength; |
| /** Count only these users jobs */ |
| Collection<String> usersIncluded; |
| /** Exclude jobs of the following users */ |
| Collection<String> usersExcluded; |
| /** Type of compression for compressed files: gzip */ |
| Class<? extends CompressionCodec> compressionClass; |
| |
| JHLAMapper() throws IOException { |
| } |
| |
| JHLAMapper(Configuration conf) throws IOException { |
| configure(new JobConf(conf)); |
| } |
| |
| public void configure(JobConf conf) { |
| super.configure(conf ); |
| usersIncluded = getUserList(conf.get("jhla.users.included", null)); |
| usersExcluded = getUserList(conf.get("jhla.users.excluded", null)); |
| String zipClassName = conf.get("jhla.compression.class", null); |
| try { |
| compressionClass = (zipClassName == null) ? null : |
| Class.forName(zipClassName).asSubclass(CompressionCodec.class); |
| } catch(Exception e) { |
| throw new RuntimeException("Compression codec not found: ", e); |
| } |
| jobDelimiterPattern = conf.get("jhla.job.delimiter.pattern", null); |
| maxJobDelimiterLineLength = conf.getInt("jhla.job.delimiter.length", 512); |
| } |
| |
| @Override |
| public void map(Text key, |
| LongWritable value, |
| OutputCollector<Text, Text> output, |
| Reporter reporter) throws IOException { |
| String name = key.toString(); |
| long longValue = value.get(); |
| |
| reporter.setStatus("starting " + name + " ::host = " + hostName); |
| |
| long tStart = System.currentTimeMillis(); |
| parseLogFile(fs, new Path(name), longValue, output, reporter); |
| long tEnd = System.currentTimeMillis(); |
| long execTime = tEnd - tStart; |
| |
| reporter.setStatus("finished " + name + " ::host = " + hostName + |
| " in " + execTime/1000 + " sec."); |
| } |
| |
| public Object doIO(Reporter reporter, |
| String path, // full path of history log file |
| long offset // starting offset within the file |
| ) throws IOException { |
| return null; |
| } |
| |
| void collectStats(OutputCollector<Text, Text> output, |
| String name, |
| long execTime, |
| Object jobObjects) throws IOException { |
| } |
| |
| private boolean isEndOfJobLog(String line) { |
| if(jobDelimiterPattern == null) |
| return false; |
| return line.matches(jobDelimiterPattern); |
| } |
| |
| /** |
| * Collect information about one job. |
| * |
| * @param fs - file system |
| * @param filePath - full path of a history log file |
| * @param offset - starting offset in the history log file |
| * @throws IOException |
| */ |
| public void parseLogFile(FileSystem fs, |
| Path filePath, |
| long offset, |
| OutputCollector<Text, Text> output, |
| Reporter reporter |
| ) throws IOException { |
| InputStream in = null; |
| try { |
| // open file & seek |
| FSDataInputStream stm = fs.open(filePath); |
| stm.seek(offset); |
| in = stm; |
| LOG.info("Opened " + filePath); |
| reporter.setStatus("Opened " + filePath); |
| // get a compression filter if specified |
| if(compressionClass != null) { |
| CompressionCodec codec = (CompressionCodec) |
| ReflectionUtils.newInstance(compressionClass, new Configuration()); |
| in = codec.createInputStream(stm); |
| LOG.info("Codec created " + filePath); |
| reporter.setStatus("Codec created " + filePath); |
| } |
| BufferedReader reader = new BufferedReader(new InputStreamReader(in)); |
| LOG.info("Reader created " + filePath); |
| // skip to the next job log start |
| long processed = 0L; |
| if(jobDelimiterPattern != null) { |
| for(String line = reader.readLine(); |
| line != null; line = reader.readLine()) { |
| if((stm.getPos() - processed) > 100000) { |
| processed = stm.getPos(); |
| reporter.setStatus("Processing " + filePath + " at " + processed); |
| } |
| if(isEndOfJobLog(line)) |
| break; |
| } |
| } |
| // parse lines and update job history |
| JobHistoryLog jh = new JobHistoryLog(); |
| int jobLineCount = 0; |
| for(String line = readLine(reader); |
| line != null; line = readLine(reader)) { |
| jobLineCount++; |
| if((stm.getPos() - processed) > 20000) { |
| processed = stm.getPos(); |
| long numTasks = (jh.tasks == null ? 0 : jh.tasks.size()); |
| String txt = "Processing " + filePath + " at " + processed |
| + " # tasks = " + numTasks; |
| reporter.setStatus(txt); |
| LOG.info(txt); |
| } |
| if(isEndOfJobLog(line)) { |
| if(jh.JOBID != null) { |
| LOG.info("Finished parsing job: " + jh.JOBID |
| + " line count = " + jobLineCount); |
| collectJobStats(jh, output, reporter); |
| LOG.info("Collected stats for job: " + jh.JOBID); |
| } |
| jh = new JobHistoryLog(); |
| jobLineCount = 0; |
| } else |
| jh.parseLine(line); |
| } |
| if(jh.JOBID == null) { |
| LOG.error("JOBID = NULL in " + filePath + " at " + processed); |
| return; |
| } |
| collectJobStats(jh, output, reporter); |
| } catch(Exception ie) { |
| // parsing errors can happen if the file has been truncated |
| LOG.error("JHLAMapper.parseLogFile", ie); |
| reporter.setStatus("JHLAMapper.parseLogFile failed " |
| + StringUtils.stringifyException(ie)); |
| throw new IOException("Job failed.", ie); |
| } finally { |
| if(in != null) in.close(); |
| } |
| } |
| |
| /** |
| * Read lines until one ends with a " ." or "\" " |
| */ |
| private StringBuffer resBuffer = new StringBuffer(); |
| private String readLine(BufferedReader reader) throws IOException { |
| resBuffer.setLength(0); |
| reader.mark(maxJobDelimiterLineLength); |
| for(String line = reader.readLine(); |
| line != null; line = reader.readLine()) { |
| if(isEndOfJobLog(line)) { |
| if(resBuffer.length() == 0) |
| resBuffer.append(line); |
| else |
| reader.reset(); |
| break; |
| } |
| if(resBuffer.length() == 0) |
| resBuffer.append(line); |
| else if(resBuffer.length() < 32000) |
| resBuffer.append(line); |
| if(line.endsWith(" .") || line.endsWith("\" ")) { |
| break; |
| } |
| reader.mark(maxJobDelimiterLineLength); |
| } |
| String result = resBuffer.length() == 0 ? null : resBuffer.toString(); |
| resBuffer.setLength(0); |
| return result; |
| } |
| |
| private void collectPerIntervalStats(OutputCollector<Text, Text> output, |
| long start, long finish, String taskType, |
| StatSeries ... stats) throws IOException { |
| long curInterval = (start / DEFAULT_TIME_INTERVAL_MSEC) |
| * DEFAULT_TIME_INTERVAL_MSEC; |
| long curTime = start; |
| long accumTime = 0; |
| while(curTime < finish) { |
| // how much of the task time belonged to current interval |
| long nextInterval = curInterval + DEFAULT_TIME_INTERVAL_MSEC; |
| long intervalTime = ((finish < nextInterval) ? |
| finish : nextInterval) - curTime; |
| IntervalKey key = new IntervalKey("", curInterval, taskType); |
| Text val = new Text(String.valueOf(intervalTime)); |
| for(StatSeries statName : stats) { |
| key.setStatName(statName.toString()); |
| output.collect(key.getTextKey(), val); |
| } |
| |
| curTime = curInterval = nextInterval; |
| accumTime += intervalTime; |
| } |
| // For the pending stat speculative attempts may intersect. |
| // Only one of them is considered pending. |
| assert accumTime == finish - start || finish < start; |
| } |
| |
| private void collectJobStats(JobHistoryLog jh, |
| OutputCollector<Text, Text> output, |
| Reporter reporter |
| ) throws IOException { |
| if(jh == null) |
| return; |
| if(jh.tasks == null) |
| return; |
| if(jh.SUBMIT_TIME <= 0) |
| throw new IOException("Job " + jh.JOBID |
| + " SUBMIT_TIME = " + jh.SUBMIT_TIME); |
| if(usersIncluded != null && !usersIncluded.contains(jh.USER)) |
| return; |
| if(usersExcluded != null && usersExcluded.contains(jh.USER)) |
| return; |
| |
| int numAttempts = 0; |
| long totalTime = 0; |
| boolean jobSuccess = jh.isSuccessful(); |
| long jobWaitTime = jh.LAUNCH_TIME - jh.SUBMIT_TIME; |
| // attemptSubmitTime is the job's SUBMIT_TIME, |
| // or the previous attempt FINISH_TIME for all subsequent attempts |
| for(TaskHistoryLog th : jh.tasks.values()) { |
| if(th.attempts == null) |
| continue; |
| // Task is successful iff both the task and the job are a "SUCCESS" |
| long attemptSubmitTime = jh.LAUNCH_TIME; |
| boolean taskSuccess = jobSuccess && th.isSuccessful(); |
| for(TaskAttemptHistoryLog tah : th.attempts.values()) { |
| // Task attempt is considered successful iff all three statuses |
| // of the attempt, the task, and the job equal "SUCCESS" |
| boolean success = taskSuccess && tah.isSuccessful(); |
| if(tah.START_TIME == 0) { |
| LOG.error("Start time 0 for task attempt " + tah.TASK_ATTEMPT_ID); |
| continue; |
| } |
| if(tah.FINISH_TIME < tah.START_TIME) { |
| LOG.error("Finish time " + tah.FINISH_TIME + " is less than " + |
| "Start time " + tah.START_TIME + " for task attempt " + |
| tah.TASK_ATTEMPT_ID); |
| tah.FINISH_TIME = tah.START_TIME; |
| } |
| |
| if(!"MAP".equals(th.TASK_TYPE) && !"REDUCE".equals(th.TASK_TYPE) && |
| !"CLEANUP".equals(th.TASK_TYPE) && !"SETUP".equals(th.TASK_TYPE)) { |
| LOG.error("Unexpected TASK_TYPE = " + th.TASK_TYPE |
| + " for attempt " + tah.TASK_ATTEMPT_ID); |
| } |
| |
| collectPerIntervalStats(output, |
| attemptSubmitTime, tah.START_TIME, th.TASK_TYPE, |
| StatSeries.STAT_LAUNCHED_PENDING_SLOT_TIME); |
| collectPerIntervalStats(output, |
| attemptSubmitTime - jobWaitTime, tah.START_TIME, th.TASK_TYPE, |
| StatSeries.STAT_SUBMIT_PENDING_SLOT_TIME); |
| if(success) |
| collectPerIntervalStats(output, |
| tah.START_TIME, tah.FINISH_TIME, th.TASK_TYPE, |
| StatSeries.STAT_ALL_SLOT_TIME); |
| else |
| collectPerIntervalStats(output, |
| tah.START_TIME, tah.FINISH_TIME, th.TASK_TYPE, |
| StatSeries.STAT_ALL_SLOT_TIME, |
| StatSeries.STAT_FAILED_SLOT_TIME); |
| totalTime += (tah.FINISH_TIME - tah.START_TIME); |
| numAttempts++; |
| if(numAttempts % 500 == 0) { |
| reporter.setStatus("Processing " + jh.JOBID + " at " + numAttempts); |
| } |
| attemptSubmitTime = tah.FINISH_TIME; |
| } |
| } |
| LOG.info("Total Maps = " + jh.TOTAL_MAPS |
| + " Reduces = " + jh.TOTAL_REDUCES); |
| LOG.info("Finished Maps = " + jh.FINISHED_MAPS |
| + " Reduces = " + jh.FINISHED_REDUCES); |
| LOG.info("numAttempts = " + numAttempts); |
| LOG.info("totalTime = " + totalTime); |
| LOG.info("averageAttemptTime = " |
| + (numAttempts==0 ? 0 : totalTime/numAttempts)); |
| LOG.info("jobTotalTime = " + (jh.FINISH_TIME <= jh.SUBMIT_TIME? 0 : |
| jh.FINISH_TIME - jh.SUBMIT_TIME)); |
| } |
| } |
| |
| public static class JHLAPartitioner implements Partitioner<Text, Text> { |
| static final int NUM_REDUCERS = 9; |
| |
| public void configure(JobConf conf) {} |
| |
| public int getPartition(Text key, Text value, int numPartitions) { |
| IntervalKey intKey = new IntervalKey(key.toString()); |
| if(intKey.statName.equals(StatSeries.STAT_ALL_SLOT_TIME.toString())) { |
| if(intKey.taskType.equals("MAP")) |
| return 0; |
| else if(intKey.taskType.equals("REDUCE")) |
| return 1; |
| } else if(intKey.statName.equals( |
| StatSeries.STAT_SUBMIT_PENDING_SLOT_TIME.toString())) { |
| if(intKey.taskType.equals("MAP")) |
| return 2; |
| else if(intKey.taskType.equals("REDUCE")) |
| return 3; |
| } else if(intKey.statName.equals( |
| StatSeries.STAT_LAUNCHED_PENDING_SLOT_TIME.toString())) { |
| if(intKey.taskType.equals("MAP")) |
| return 4; |
| else if(intKey.taskType.equals("REDUCE")) |
| return 5; |
| } else if(intKey.statName.equals( |
| StatSeries.STAT_FAILED_SLOT_TIME.toString())) { |
| if(intKey.taskType.equals("MAP")) |
| return 6; |
| else if(intKey.taskType.equals("REDUCE")) |
| return 7; |
| } |
| return 8; |
| } |
| } |
| |
| private static void runJHLA( |
| Class<? extends Mapper<Text, LongWritable, Text, Text>> mapperClass, |
| Path outputDir, |
| Configuration fsConfig) throws IOException { |
| JobConf job = new JobConf(fsConfig, JHLogAnalyzer.class); |
| |
| job.setPartitionerClass(JHLAPartitioner.class); |
| |
| FileInputFormat.setInputPaths(job, INPUT_DIR); |
| job.setInputFormat(SequenceFileInputFormat.class); |
| |
| job.setMapperClass(mapperClass); |
| job.setReducerClass(AccumulatingReducer.class); |
| |
| FileOutputFormat.setOutputPath(job, outputDir); |
| job.setOutputKeyClass(Text.class); |
| job.setOutputValueClass(Text.class); |
| job.setNumReduceTasks(JHLAPartitioner.NUM_REDUCERS); |
| JobClient.runJob(job); |
| } |
| |
| private static class LoggingCollector implements OutputCollector<Text, Text> { |
| public void collect(Text key, Text value) throws IOException { |
| LOG.info(key + " == " + value); |
| } |
| } |
| |
| /** |
| * Run job history log analyser. |
| */ |
| public static void main(String[] args) { |
| Path resFileName = RESULT_FILE; |
| Configuration conf = new Configuration(); |
| |
| try { |
| conf.setInt("test.io.file.buffer.size", 0); |
| Path historyDir = DEFAULT_HISTORY_DIR; |
| String testFile = null; |
| boolean cleanup = false; |
| |
| boolean initControlFiles = true; |
| for (int i = 0; i < args.length; i++) { // parse command line |
| if (args[i].equalsIgnoreCase("-historyDir")) { |
| historyDir = new Path(args[++i]); |
| } else if (args[i].equalsIgnoreCase("-resFile")) { |
| resFileName = new Path(args[++i]); |
| } else if (args[i].equalsIgnoreCase("-usersIncluded")) { |
| conf.set("jhla.users.included", args[++i]); |
| } else if (args[i].equalsIgnoreCase("-usersExcluded")) { |
| conf.set("jhla.users.excluded", args[++i]); |
| } else if (args[i].equalsIgnoreCase("-gzip")) { |
| conf.set("jhla.compression.class", GzipCodec.class.getCanonicalName()); |
| } else if (args[i].equalsIgnoreCase("-jobDelimiter")) { |
| conf.set("jhla.job.delimiter.pattern", args[++i]); |
| } else if (args[i].equalsIgnoreCase("-jobDelimiterLength")) { |
| conf.setInt("jhla.job.delimiter.length", Integer.parseInt(args[++i])); |
| } else if(args[i].equalsIgnoreCase("-noInit")) { |
| initControlFiles = false; |
| } else if(args[i].equalsIgnoreCase("-test")) { |
| testFile = args[++i]; |
| } else if(args[i].equalsIgnoreCase("-clean")) { |
| cleanup = true; |
| } else if(args[i].equalsIgnoreCase("-jobQueue")) { |
| conf.set("mapred.job.queue.name", args[++i]); |
| } else if(args[i].startsWith("-Xmx")) { |
| conf.set("mapred.child.java.opts", args[i]); |
| } else { |
| printUsage(); |
| } |
| } |
| |
| if(cleanup) { |
| cleanup(conf); |
| return; |
| } |
| if(testFile != null) { |
| LOG.info("Start JHLA test ============ "); |
| LocalFileSystem lfs = FileSystem.getLocal(conf); |
| conf.set("fs.default.name", "file:///"); |
| JHLAMapper map = new JHLAMapper(conf); |
| map.parseLogFile(lfs, new Path(testFile), 0L, |
| new LoggingCollector(), Reporter.NULL); |
| return; |
| } |
| |
| FileSystem fs = FileSystem.get(conf); |
| if(initControlFiles) |
| createControlFile(fs, historyDir); |
| long tStart = System.currentTimeMillis(); |
| runJHLA(JHLAMapper.class, OUTPUT_DIR, conf); |
| long execTime = System.currentTimeMillis() - tStart; |
| |
| analyzeResult(fs, 0, execTime, resFileName); |
| } catch(IOException e) { |
| System.err.print(StringUtils.stringifyException(e)); |
| System.exit(-1); |
| } |
| } |
| |
| |
| private static void printUsage() { |
| String className = JHLogAnalyzer.class.getSimpleName(); |
| System.err.println("Usage: " + className |
| + "\n\t[-historyDir inputDir] | [-resFile resultFile] |" |
| + "\n\t[-usersIncluded | -usersExcluded userList] |" |
| + "\n\t[-gzip] | [-jobDelimiter pattern] |" |
| + "\n\t[-help | -clean | -test testFile]"); |
| System.exit(-1); |
| } |
| |
| private static Collection<String> getUserList(String users) { |
| if(users == null) |
| return null; |
| StringTokenizer tokens = new StringTokenizer(users, ",;"); |
| Collection<String> userList = new ArrayList<String>(tokens.countTokens()); |
| while(tokens.hasMoreTokens()) |
| userList.add(tokens.nextToken()); |
| return userList; |
| } |
| |
| /** |
| * Result is combined from all reduce output files and is written to |
| * RESULT_FILE in the format |
| * column 1: |
| */ |
| private static void analyzeResult( FileSystem fs, |
| int testType, |
| long execTime, |
| Path resFileName |
| ) throws IOException { |
| LOG.info("Analizing results ..."); |
| DataOutputStream out = null; |
| BufferedWriter writer = null; |
| try { |
| out = new DataOutputStream(fs.create(resFileName)); |
| writer = new BufferedWriter(new OutputStreamWriter(out)); |
| writer.write("SERIES\tPERIOD\tTYPE\tSLOT_HOUR\n"); |
| FileStatus[] reduceFiles = fs.listStatus(OUTPUT_DIR); |
| assert reduceFiles.length == JHLAPartitioner.NUM_REDUCERS; |
| for(int i = 0; i < JHLAPartitioner.NUM_REDUCERS; i++) { |
| DataInputStream in = null; |
| BufferedReader lines = null; |
| try { |
| in = fs.open(reduceFiles[i].getPath()); |
| lines = new BufferedReader(new InputStreamReader(in)); |
| |
| String line; |
| while((line = lines.readLine()) != null) { |
| StringTokenizer tokens = new StringTokenizer(line, "\t*"); |
| String attr = tokens.nextToken(); |
| String dateTime = tokens.nextToken(); |
| String taskType = tokens.nextToken(); |
| double val = Long.parseLong(tokens.nextToken()) / |
| (double)DEFAULT_TIME_INTERVAL_MSEC; |
| writer.write(attr.substring(2)); // skip the stat type "l:" |
| writer.write("\t"); |
| writer.write(dateTime); |
| writer.write("\t"); |
| writer.write(taskType); |
| writer.write("\t"); |
| writer.write(String.valueOf((float)val)); |
| writer.newLine(); |
| } |
| } finally { |
| if(lines != null) lines.close(); |
| if(in != null) in.close(); |
| } |
| } |
| } finally { |
| if(writer != null) writer.close(); |
| if(out != null) out.close(); |
| } |
| LOG.info("Analizing results ... done."); |
| } |
| |
| private static void cleanup(Configuration conf) throws IOException { |
| LOG.info("Cleaning up test files"); |
| FileSystem fs = FileSystem.get(conf); |
| fs.delete(new Path(JHLA_ROOT_DIR), true); |
| } |
| } |
