|
|
@@ -0,0 +1,1129 @@
|
|
|
+/**
|
|
|
+ * 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);
|
|
|
+ }
|
|
|
+}
|
Konstantin Shvachko