();
+ for (CounterName cn : CounterName.values()) {
+ _counterDisplayNameMap.put(cn._displayName, cn);
+ _counterNameMap.put(cn._name, cn);
+ }
+ }
+
+ public static CounterName getCounterFromName(String name) {
+ if (_counterNameMap.containsKey(name)) {
+ return _counterNameMap.get(name);
+ }
+ return null;
+ }
+
+ public static CounterName getCounterFromDisplayName(String displayName) {
+ if (_counterDisplayNameMap.containsKey(displayName)) {
+ return _counterDisplayNameMap.get(displayName);
+ }
+ return null;
+ }
+
+ public String getName() {
+ return _name;
+ }
+
+ public String getDisplayName() {
+ return _displayName;
+ }
+
+ public String getGroupName() {
+ return _group.name();
+ }
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/data/MapReduceTaskData.java b/app/com/linkedin/drelephant/mapreduce/data/MapReduceTaskData.java
new file mode 100644
index 000000000..cef9fd44c
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/data/MapReduceTaskData.java
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.data;
+
+
+/**
+ * This class manages the MapReduce Tasks
+ */
+public class MapReduceTaskData {
+ private MapReduceCounterData _counterHolder;
+ private String _taskId;
+ // The successful attempt id
+ private String _attemptId;
+ private long _totalTimeMs = 0;
+ private long _shuffleTimeMs = 0;
+ private long _sortTimeMs = 0;
+ private long _startTimeMs = 0;
+ private long _finishTimeMs = 0;
+ private boolean _sampled = false;
+
+ public MapReduceTaskData(MapReduceCounterData counterHolder, long[] time) {
+ this._counterHolder = counterHolder;
+ this._totalTimeMs = time[0];
+ this._shuffleTimeMs = time[1];
+ this._sortTimeMs = time[2];
+ this._startTimeMs = time[3];
+ this._finishTimeMs = time[4];
+ this._sampled = true;
+ }
+
+ public MapReduceTaskData(MapReduceCounterData counterHolder) {
+ this._counterHolder = counterHolder;
+ }
+
+ public MapReduceTaskData(String taskId, String taskAttemptId) {
+ this._taskId = taskId;
+ this._attemptId = taskAttemptId;
+ }
+
+ public void setCounter(MapReduceCounterData counterHolder) {
+ this._counterHolder = counterHolder;
+ this._sampled = true;
+ }
+
+ public void setTime(long[] time) {
+ this._totalTimeMs = time[0];
+ this._shuffleTimeMs = time[1];
+ this._sortTimeMs = time[2];
+ this._startTimeMs = time[3];
+ this._finishTimeMs = time[4];
+ this._sampled = true;
+ }
+
+ public MapReduceCounterData getCounters() {
+ return _counterHolder;
+ }
+
+ public long getTotalRunTimeMs() {
+ return _totalTimeMs;
+ }
+
+ public long getCodeExecutionTimeMs() {
+ return _totalTimeMs - _shuffleTimeMs - _sortTimeMs;
+ }
+
+ public long getShuffleTimeMs() {
+ return _shuffleTimeMs;
+ }
+
+ public long getSortTimeMs() {
+ return _sortTimeMs;
+ }
+
+ public long getStartTimeMs() {
+ return _startTimeMs;
+ }
+
+ public long getFinishTimeMs() {
+ return _finishTimeMs;
+ }
+
+ public boolean isSampled() {
+ return _sampled;
+ }
+
+ public String getTaskId() {
+ return _taskId;
+ }
+
+ public String getAttemptId() {
+ return _attemptId;
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFSFetcherHadoop2.java b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFSFetcherHadoop2.java
new file mode 100644
index 000000000..19e43e512
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFSFetcherHadoop2.java
@@ -0,0 +1,344 @@
+/*
+ * Copyright 2016 Linkin Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.fetchers;
+
+import com.linkedin.drelephant.analysis.AnalyticJob;
+import com.linkedin.drelephant.configurations.fetcher.FetcherConfigurationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.util.Utils;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.mapreduce.Counter;
+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;
+import org.apache.log4j.Logger;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * This class implements the Fetcher for MapReduce Applications on Hadoop2
+ * Instead of fetching data from job history server, it retrieves history logs and job configs from
+ * HDFS directly. Each job's data consists of a JSON event log file with extension ".jhist" and an
+ * XML job configuration file.
+ */
+public class MapReduceFSFetcherHadoop2 extends MapReduceFetcher {
+ private static final Logger logger = Logger.getLogger(MapReduceFSFetcherHadoop2.class);
+
+ private static final String LOG_SIZE_XML_FIELD = "history_log_size_limit_in_mb";
+ private static final String TIMESTAMP_DIR_FORMAT = "%04d" + File.separator + "%02d" + File.separator + "%02d";
+ private static final int SERIAL_NUMBER_DIRECTORY_DIGITS = 6;
+ protected static final double DEFALUT_MAX_LOG_SIZE_IN_MB = 500;
+
+ private FileSystem _fs;
+ private String _historyLocation;
+ private String _intermediateHistoryLocation;
+ private double _maxLogSizeInMB;
+
+ public MapReduceFSFetcherHadoop2(FetcherConfigurationData fetcherConfData) throws IOException {
+ super(fetcherConfData);
+
+ _maxLogSizeInMB = DEFALUT_MAX_LOG_SIZE_IN_MB;
+ if (fetcherConfData.getParamMap().get(LOG_SIZE_XML_FIELD) != null) {
+ double[] logLimitSize = Utils.getParam(fetcherConfData.getParamMap().get(LOG_SIZE_XML_FIELD), 1);
+ if (logLimitSize != null) {
+ _maxLogSizeInMB = logLimitSize[0];
+ }
+ }
+ logger.info("The history log limit of MapReduce application is set to " + _maxLogSizeInMB + " MB");
+
+ Configuration conf = new Configuration();
+ this._fs = FileSystem.get(conf);
+ this._historyLocation = conf.get("mapreduce.jobhistory.done-dir");
+ this._intermediateHistoryLocation = conf.get("mapreduce.jobhistory.intermediate-done-dir");
+ logger.info("Intermediate history dir: " + _intermediateHistoryLocation);
+ logger.info("History done dir: " + _historyLocation);
+ }
+
+ public String getHistoryLocation() {
+ return _historyLocation;
+ }
+
+ public double getMaxLogSizeInMB() {
+ return _maxLogSizeInMB;
+ }
+
+ /**
+ * The location of a job history file is in format: {done-dir}/yyyy/mm/dd/{serialPart}.
+ * yyyy/mm/dd is the year, month and date of the finish time.
+ * serialPart is the first 6 digits of the serial number considering it as a 9 digits number.
+ * PS: The serial number is the last part of an app id.
+ *
+ * For example, if appId = application_1461566847127_84624, then serial number is 84624.
+ * Consider it as a 9 digits number, serial number is 000084624. So the corresponding
+ * serialPart is 000084. If this application finish at 2016-5-30, its history file will locate
+ * at {done-dir}/2016/05/30/000084
+ *
+ *
+ * Furthermore, this location format is only satisfied for finished jobs in {done-dir} and not
+ * for running jobs in {intermediate-done-dir}.
+ *
+ */
+ protected String getHistoryDir(AnalyticJob job) {
+ // generate the date part
+ Calendar timestamp = Calendar.getInstance();
+ timestamp.setTimeInMillis(job.getFinishTime());
+ String datePart = String.format(TIMESTAMP_DIR_FORMAT,
+ timestamp.get(Calendar.YEAR),
+ timestamp.get(Calendar.MONTH) + 1,
+ timestamp.get(Calendar.DAY_OF_MONTH));
+
+ // generate the serial part
+ String appId = job.getAppId();
+ int serialNumber = Integer.parseInt(appId.substring(appId.lastIndexOf('_') + 1));
+ String serialPart = String.format("%09d", serialNumber)
+ .substring(0, SERIAL_NUMBER_DIRECTORY_DIGITS);
+
+ return StringUtils.join(new String[]{_historyLocation, datePart, serialPart, ""}, File.separator);
+ }
+
+ private DataFiles getHistoryFiles(AnalyticJob job) throws IOException {
+ String jobId = Utils.getJobIdFromApplicationId(job.getAppId());
+ String jobConfPath = null;
+ String jobHistPath = null;
+
+ // Search files in done dir
+ String jobHistoryDirPath = getHistoryDir(job);
+ RemoteIterator it = _fs.listFiles(new Path(jobHistoryDirPath), false);
+ while (it.hasNext() && (jobConfPath == null || jobHistPath == null)) {
+ String name = it.next().getPath().getName();
+ if (name.contains(jobId)) {
+ if (name.endsWith("_conf.xml")) {
+ jobConfPath = jobHistoryDirPath + name;
+ } else if (name.endsWith(".jhist")) {
+ jobHistPath = jobHistoryDirPath + name;
+ }
+ }
+ }
+
+ // If some files are missing, search in the intermediate-done-dir in case the HistoryServer has
+ // not yet moved them into the done-dir.
+ String intermediateDirPath = _intermediateHistoryLocation + File.separator + job.getUser() + File.separator;
+ if (jobConfPath == null) {
+ jobConfPath = intermediateDirPath + jobId + "_conf.xml";
+ if (!_fs.exists(new Path(jobConfPath))) {
+ throw new FileNotFoundException("Can't find config of " + jobId + " in neither "
+ + jobHistoryDirPath + " nor " + intermediateDirPath);
+ }
+ logger.info("Found job config in intermediate dir: " + jobConfPath);
+ }
+ if (jobHistPath == null) {
+ try {
+ it = _fs.listFiles(new Path(intermediateDirPath), false);
+ while (it.hasNext()) {
+ String name = it.next().getPath().getName();
+ if (name.contains(jobId) && name.endsWith(".jhist")) {
+ jobHistPath = intermediateDirPath + name;
+ logger.info("Found history file in intermediate dir: " + jobHistPath);
+ break;
+ }
+ }
+ } catch (FileNotFoundException e) {
+ logger.error("Intermediate history directory " + intermediateDirPath + " not found");
+ }
+ if (jobHistPath == null) {
+ throw new FileNotFoundException("Can't find history file of " + jobId + " in neither "
+ + jobHistoryDirPath + " nor " + intermediateDirPath);
+ }
+ }
+
+ return new DataFiles(jobConfPath, jobHistPath);
+ }
+
+ @Override
+ public MapReduceApplicationData fetchData(AnalyticJob job) throws IOException {
+ DataFiles files = getHistoryFiles(job);
+ String confFile = files.getJobConfPath();
+ String histFile = files.getJobHistPath();
+ String appId = job.getAppId();
+ String jobId = Utils.getJobIdFromApplicationId(appId);
+
+ MapReduceApplicationData jobData = new MapReduceApplicationData();
+ jobData.setAppId(appId).setJobId(jobId);
+
+ // Fetch job config
+ Configuration jobConf = new Configuration(false);
+ jobConf.addResource(_fs.open(new Path(confFile)), confFile);
+ Properties jobConfProperties = new Properties();
+ for (Map.Entry entry : jobConf) {
+ jobConfProperties.put(entry.getKey(), entry.getValue());
+ }
+ jobData.setJobConf(jobConfProperties);
+
+ // Check if job history file is too large and should be throttled
+ if (_fs.getFileStatus(new Path(histFile)).getLen() > _maxLogSizeInMB * FileUtils.ONE_MB) {
+ String errMsg = "The history log of MapReduce application: " + appId + " is over the limit size of "
+ + _maxLogSizeInMB + " MB, the parsing process gets throttled.";
+ logger.warn(errMsg);
+ jobData.setDiagnosticInfo(errMsg);
+ jobData.setSucceeded(false); // set succeeded to false to avoid heuristic analysis
+ return jobData;
+ }
+
+ // Analyze job history file
+ JobHistoryParser parser = new JobHistoryParser(_fs, histFile);
+ JobHistoryParser.JobInfo jobInfo = parser.parse();
+ IOException parseException = parser.getParseException();
+ if (parseException != null) {
+ throw new RuntimeException("Could not parse history file " + histFile, parseException);
+ }
+
+ jobData.setSubmitTime(jobInfo.getSubmitTime());
+ jobData.setStartTime(jobInfo.getLaunchTime());
+ jobData.setFinishTime(jobInfo.getFinishTime());
+
+ String state = jobInfo.getJobStatus();
+ if (state.equals("SUCCEEDED")) {
+
+ jobData.setSucceeded(true);
+
+ // Fetch job counter
+ MapReduceCounterData jobCounter = getCounterData(jobInfo.getTotalCounters());
+
+ // Fetch task data
+ Map allTasks = jobInfo.getAllTasks();
+ List mapperInfoList = new ArrayList();
+ List reducerInfoList = new ArrayList();
+ for (JobHistoryParser.TaskInfo taskInfo : allTasks.values()) {
+ if (taskInfo.getTaskType() == TaskType.MAP) {
+ mapperInfoList.add(taskInfo);
+ } else {
+ reducerInfoList.add(taskInfo);
+ }
+ }
+ if (jobInfo.getTotalMaps() > MAX_SAMPLE_SIZE) {
+ logger.debug(jobId + " total mappers: " + mapperInfoList.size());
+ }
+ if (jobInfo.getTotalReduces() > MAX_SAMPLE_SIZE) {
+ logger.debug(jobId + " total reducers: " + reducerInfoList.size());
+ }
+ MapReduceTaskData[] mapperList = getTaskData(jobId, mapperInfoList);
+ MapReduceTaskData[] reducerList = getTaskData(jobId, reducerInfoList);
+
+ jobData.setCounters(jobCounter).setMapperData(mapperList).setReducerData(reducerList);
+ } else if (state.equals("FAILED")) {
+
+ jobData.setSucceeded(false);
+ jobData.setDiagnosticInfo(jobInfo.getErrorInfo());
+ } else {
+ // Should not reach here
+ throw new RuntimeException("Job state not supported. Should be either SUCCEEDED or FAILED");
+ }
+
+ return jobData;
+ }
+
+ private MapReduceCounterData getCounterData(Counters counters) {
+ MapReduceCounterData holder = new MapReduceCounterData();
+ for (CounterGroup group : counters) {
+ String groupName = group.getName();
+ for (Counter counter : group) {
+ holder.set(groupName, counter.getName(), counter.getValue());
+ }
+ }
+ return holder;
+ }
+
+ private long[] getTaskExecTime(JobHistoryParser.TaskAttemptInfo attempInfo) {
+ long startTime = attempInfo.getStartTime();
+ long finishTime = attempInfo.getFinishTime();
+ boolean isMapper = (attempInfo.getTaskType() == TaskType.MAP);
+
+ long[] time;
+ if (isMapper) {
+ time = new long[]{finishTime - startTime, 0, 0, startTime, finishTime};
+ } else {
+ long shuffleFinishTime = attempInfo.getShuffleFinishTime();
+ long mergeFinishTime = attempInfo.getSortFinishTime();
+ time = new long[]{finishTime - startTime, shuffleFinishTime - startTime,
+ mergeFinishTime - shuffleFinishTime, startTime, finishTime};
+ }
+ return time;
+ }
+
+ private MapReduceTaskData[] getTaskData(String jobId, List infoList) {
+ int sampleSize = sampleAndGetSize(jobId, infoList);
+
+ MapReduceTaskData[] taskList = new MapReduceTaskData[sampleSize];
+ for (int i = 0; i < sampleSize; i++) {
+ JobHistoryParser.TaskInfo tInfo = infoList.get(i);
+ if (!"SUCCEEDED".equals(tInfo.getTaskStatus())) {
+ System.out.println("This is a failed task: " + tInfo.getTaskId().toString());
+ continue;
+ }
+
+ String taskId = tInfo.getTaskId().toString();
+ TaskAttemptID attemptId = tInfo.getSuccessfulAttemptId();
+ taskList[i] = new MapReduceTaskData(taskId, attemptId.toString());
+
+ MapReduceCounterData taskCounterData = getCounterData(tInfo.getCounters());
+ long[] taskExecTime = getTaskExecTime(tInfo.getAllTaskAttempts().get(attemptId));
+
+ taskList[i].setCounter(taskCounterData);
+ taskList[i].setTime(taskExecTime);
+ }
+ return taskList;
+ }
+
+ private class DataFiles {
+ private String jobConfPath;
+ private String jobHistPath;
+
+ public DataFiles(String confPath, String histPath) {
+ this.jobConfPath = confPath;
+ this.jobHistPath = histPath;
+ }
+
+ public String getJobConfPath() {
+ return jobConfPath;
+ }
+
+ public void setJobConfPath(String jobConfPath) {
+ this.jobConfPath = jobConfPath;
+ }
+
+ public String getJobHistPath() {
+ return jobHistPath;
+ }
+
+ public void setJobHistPath(String jobHistPath) {
+ this.jobHistPath = jobHistPath;
+ }
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcher.java b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcher.java
new file mode 100644
index 000000000..83b7aef4f
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcher.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2016 Linkin Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.fetchers;
+
+import com.linkedin.drelephant.analysis.ElephantFetcher;
+import com.linkedin.drelephant.configurations.fetcher.FetcherConfigurationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import org.apache.log4j.Logger;
+
+import java.util.Collections;
+import java.util.List;
+
+
+public abstract class MapReduceFetcher implements ElephantFetcher {
+ private static final Logger logger = Logger.getLogger(MapReduceFetcher.class);
+ protected static final int MAX_SAMPLE_SIZE = 200;
+ protected static final String SAMPLING_ENABLED_XML_FIELD = "sampling_enabled";
+
+ protected FetcherConfigurationData _fetcherConfigurationData;
+ private boolean _samplingEnabled;
+
+ public MapReduceFetcher(FetcherConfigurationData fetcherConfData) {
+ this._fetcherConfigurationData = fetcherConfData;
+ this._samplingEnabled = Boolean.parseBoolean(
+ fetcherConfData.getParamMap().get(SAMPLING_ENABLED_XML_FIELD));
+ }
+
+ protected int sampleAndGetSize(String jobId, List taskList) {
+ // check if sampling is enabled
+ if (_samplingEnabled) {
+ if (taskList.size() > MAX_SAMPLE_SIZE) {
+ logger.info(jobId + " needs sampling.");
+ Collections.shuffle(taskList);
+ }
+ return Math.min(taskList.size(), MAX_SAMPLE_SIZE);
+ }
+ return taskList.size();
+ }
+
+ public boolean isSamplingEnabled() {
+ return _samplingEnabled;
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcherHadoop2.java b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcherHadoop2.java
new file mode 100644
index 000000000..6dffbd0ae
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcherHadoop2.java
@@ -0,0 +1,437 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.fetchers;
+
+import com.linkedin.drelephant.analysis.AnalyticJob;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.math.Statistics;
+import com.linkedin.drelephant.configurations.fetcher.FetcherConfigurationData;
+import com.linkedin.drelephant.util.Utils;
+
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.authentication.client.AuthenticatedURL;
+import org.apache.hadoop.security.authentication.client.AuthenticationException;
+import org.apache.log4j.Logger;
+import org.codehaus.jackson.JsonNode;
+import org.codehaus.jackson.map.ObjectMapper;
+
+
+/**
+ * This class implements the Fetcher for MapReduce Applications on Hadoop2
+ */
+public class MapReduceFetcherHadoop2 extends MapReduceFetcher {
+ private static final Logger logger = Logger.getLogger(MapReduceFetcherHadoop2.class);
+ // We provide one minute job fetch delay due to the job sending lag from AM/NM to JobHistoryServer HDFS
+
+ private URLFactory _urlFactory;
+ private JSONFactory _jsonFactory;
+ private String _jhistoryWebAddr;
+
+ public MapReduceFetcherHadoop2(FetcherConfigurationData fetcherConfData) throws IOException {
+ super(fetcherConfData);
+
+ final String jhistoryAddr = new Configuration().get("mapreduce.jobhistory.webapp.address");
+
+ logger.info("Connecting to the job history server at " + jhistoryAddr + "...");
+ _urlFactory = new URLFactory(jhistoryAddr);
+ logger.info("Connection success.");
+
+ _jsonFactory = new JSONFactory();
+ _jhistoryWebAddr = "http://" + jhistoryAddr + "/jobhistory/job/";
+ }
+
+ @Override
+ public MapReduceApplicationData fetchData(AnalyticJob analyticJob) throws IOException, AuthenticationException {
+ String appId = analyticJob.getAppId();
+ MapReduceApplicationData jobData = new MapReduceApplicationData();
+ String jobId = Utils.getJobIdFromApplicationId(appId);
+ jobData.setAppId(appId).setJobId(jobId);
+ // Change job tracking url to job history page
+ analyticJob.setTrackingUrl(_jhistoryWebAddr + jobId);
+ try {
+
+ // Fetch job config
+ Properties jobConf = _jsonFactory.getProperties(_urlFactory.getJobConfigURL(jobId));
+ jobData.setJobConf(jobConf);
+
+ URL jobURL = _urlFactory.getJobURL(jobId);
+ String state = _jsonFactory.getState(jobURL);
+
+ jobData.setSubmitTime(_jsonFactory.getSubmitTime(jobURL));
+ jobData.setStartTime(_jsonFactory.getStartTime(jobURL));
+ jobData.setFinishTime(_jsonFactory.getFinishTime(jobURL));
+
+ if (state.equals("SUCCEEDED")) {
+
+ jobData.setSucceeded(true);
+
+ // Fetch job counter
+ MapReduceCounterData jobCounter = _jsonFactory.getJobCounter(_urlFactory.getJobCounterURL(jobId));
+
+ // Fetch task data
+ URL taskListURL = _urlFactory.getTaskListURL(jobId);
+ List mapperList = new ArrayList();
+ List reducerList = new ArrayList();
+ _jsonFactory.getTaskDataAll(taskListURL, jobId, mapperList, reducerList);
+
+ MapReduceTaskData[] mapperData = mapperList.toArray(new MapReduceTaskData[mapperList.size()]);
+ MapReduceTaskData[] reducerData = reducerList.toArray(new MapReduceTaskData[reducerList.size()]);
+
+ jobData.setCounters(jobCounter).setMapperData(mapperData).setReducerData(reducerData);
+ } else if (state.equals("FAILED")) {
+
+ jobData.setSucceeded(false);
+ String diagnosticInfo;
+ try {
+ diagnosticInfo = parseException(jobData.getJobId(), _jsonFactory.getDiagnosticInfo(jobURL));
+ } catch(Exception e) {
+ diagnosticInfo = null;
+ }
+ jobData.setDiagnosticInfo(diagnosticInfo);
+ } else {
+ // Should not reach here
+ throw new RuntimeException("Job state not supported. Should be either SUCCEEDED or FAILED");
+ }
+ } finally {
+ ThreadContextMR2.updateAuthToken();
+ }
+
+ return jobData;
+ }
+
+ private String parseException(String jobId, String diagnosticInfo) throws MalformedURLException, IOException,
+ AuthenticationException {
+ Matcher m = ThreadContextMR2.getDiagnosticMatcher(diagnosticInfo);
+ if (m.matches()) {
+ if (Integer.parseInt(m.group(2)) == 0) {
+ // This is due to bug in hadoop 2.3 and shoufixed in 2.4
+ throw new RuntimeException("Error in diagnosticInfo");
+ }
+ String taskId = m.group(1);
+ System.out.println("parse succedded. " + m.group(1) + " " + m.group(2));
+ return _jsonFactory.getTaskFailedStackTrace(_urlFactory.getTaskAllAttemptsURL(jobId, taskId));
+ }
+ logger.info("Does not match regex!!");
+ // Diagnostic info not present in the job. Usually due to exception during AM setup
+ throw new RuntimeException("No sufficient diagnostic Info");
+ }
+
+ private URL getTaskCounterURL(String jobId, String taskId) throws MalformedURLException {
+ return _urlFactory.getTaskCounterURL(jobId, taskId);
+ }
+
+ private URL getTaskAttemptURL(String jobId, String taskId, String attemptId) throws MalformedURLException {
+ return _urlFactory.getTaskAttemptURL(jobId, taskId, attemptId);
+ }
+
+ private class URLFactory {
+
+ private String _restRoot;
+
+ private URLFactory(String hserverAddr) throws IOException {
+ _restRoot = "http://" + hserverAddr + "/ws/v1/history/mapreduce/jobs";
+ verifyURL(_restRoot);
+ }
+
+ private void verifyURL(String url) throws IOException {
+ final URLConnection connection = new URL(url).openConnection();
+ // Check service availability
+ connection.connect();
+ return;
+ }
+
+ private URL getJobURL(String jobId) throws MalformedURLException {
+ return new URL(_restRoot + "/" + jobId);
+ }
+
+ private URL getJobConfigURL(String jobId) throws MalformedURLException {
+ return new URL(_restRoot + "/" + jobId + "/conf");
+ }
+
+ private URL getJobCounterURL(String jobId) throws MalformedURLException {
+ return new URL(_restRoot + "/" + jobId + "/counters");
+ }
+
+ private URL getTaskListURL(String jobId) throws MalformedURLException {
+ return new URL(_restRoot + "/" + jobId + "/tasks");
+ }
+
+ private URL getTaskCounterURL(String jobId, String taskId) throws MalformedURLException {
+ return new URL(_restRoot + "/" + jobId + "/tasks/" + taskId + "/counters");
+ }
+
+ private URL getTaskAllAttemptsURL(String jobId, String taskId) throws MalformedURLException {
+ return new URL(_restRoot + "/" + jobId + "/tasks/" + taskId + "/attempts");
+ }
+
+ private URL getTaskAttemptURL(String jobId, String taskId, String attemptId) throws MalformedURLException {
+ return new URL(_restRoot + "/" + jobId + "/tasks/" + taskId + "/attempts/" + attemptId);
+ }
+ }
+
+ private class JSONFactory {
+
+ private long getStartTime(URL url) throws IOException, AuthenticationException {
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ return rootNode.path("job").path("startTime").getValueAsLong();
+ }
+
+ private long getFinishTime(URL url) throws IOException, AuthenticationException {
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ return rootNode.path("job").path("finishTime").getValueAsLong();
+ }
+
+ private long getSubmitTime(URL url) throws IOException, AuthenticationException {
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ return rootNode.path("job").path("submitTime").getValueAsLong();
+ }
+
+ private String getState(URL url) throws IOException, AuthenticationException {
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ return rootNode.path("job").path("state").getValueAsText();
+ }
+
+ private String getDiagnosticInfo(URL url) throws IOException, AuthenticationException {
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ String diag = rootNode.path("job").path("diagnostics").getValueAsText();
+ return diag;
+ }
+
+ private Properties getProperties(URL url) throws IOException, AuthenticationException {
+ Properties jobConf = new Properties();
+
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ JsonNode configs = rootNode.path("conf").path("property");
+
+ for (JsonNode conf : configs) {
+ String key = conf.get("name").getValueAsText();
+ String val = conf.get("value").getValueAsText();
+ jobConf.setProperty(key, val);
+ }
+ return jobConf;
+ }
+
+ private MapReduceCounterData getJobCounter(URL url) throws IOException, AuthenticationException {
+ MapReduceCounterData holder = new MapReduceCounterData();
+
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ JsonNode groups = rootNode.path("jobCounters").path("counterGroup");
+
+ for (JsonNode group : groups) {
+ for (JsonNode counter : group.path("counter")) {
+ String counterName = counter.get("name").getValueAsText();
+ Long counterValue = counter.get("totalCounterValue").getLongValue();
+ String groupName = group.get("counterGroupName").getValueAsText();
+ holder.set(groupName, counterName, counterValue);
+ }
+ }
+ return holder;
+ }
+
+ private MapReduceCounterData getTaskCounter(URL url) throws IOException, AuthenticationException {
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ JsonNode groups = rootNode.path("jobTaskCounters").path("taskCounterGroup");
+ MapReduceCounterData holder = new MapReduceCounterData();
+
+ for (JsonNode group : groups) {
+ for (JsonNode counter : group.path("counter")) {
+ String name = counter.get("name").getValueAsText();
+ String groupName = group.get("counterGroupName").getValueAsText();
+ Long value = counter.get("value").getLongValue();
+ holder.set(groupName, name, value);
+ }
+ }
+ return holder;
+ }
+
+ private long[] getTaskExecTime(URL url) throws IOException, AuthenticationException {
+
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ JsonNode taskAttempt = rootNode.path("taskAttempt");
+
+ long startTime = taskAttempt.get("startTime").getLongValue();
+ long finishTime = taskAttempt.get("finishTime").getLongValue();
+ boolean isMapper = taskAttempt.get("type").getValueAsText().equals("MAP");
+
+ long[] time;
+ if (isMapper) {
+ // No shuffle sore time in Mapper
+ time = new long[] { finishTime - startTime, 0, 0 ,startTime, finishTime};
+ } else {
+ long shuffleTime = taskAttempt.get("elapsedShuffleTime").getLongValue();
+ long sortTime = taskAttempt.get("elapsedMergeTime").getLongValue();
+ time = new long[] { finishTime - startTime, shuffleTime, sortTime, startTime, finishTime };
+ }
+
+ return time;
+ }
+
+ private void getTaskDataAll(URL url, String jobId, List mapperList,
+ List reducerList) throws IOException, AuthenticationException {
+
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(url);
+ JsonNode tasks = rootNode.path("tasks").path("task");
+
+ for (JsonNode task : tasks) {
+ String state = task.get("state").getValueAsText();
+ if (!state.equals("SUCCEEDED")) {
+ // This is a failed task.
+ continue;
+ }
+ String taskId = task.get("id").getValueAsText();
+ String attemptId = task.get("successfulAttempt").getValueAsText();
+ boolean isMapper = task.get("type").getValueAsText().equals("MAP");
+
+ if (isMapper) {
+ mapperList.add(new MapReduceTaskData(taskId, attemptId));
+ } else {
+ reducerList.add(new MapReduceTaskData(taskId, attemptId));
+ }
+ }
+
+ getTaskData(jobId, mapperList);
+ getTaskData(jobId, reducerList);
+ }
+
+ private void getTaskData(String jobId, List taskList) throws IOException, AuthenticationException {
+
+ int sampleSize = sampleAndGetSize(jobId, taskList);
+
+ for(int i=0; i < sampleSize; i++) {
+ MapReduceTaskData data = taskList.get(i);
+
+ URL taskCounterURL = getTaskCounterURL(jobId, data.getTaskId());
+ MapReduceCounterData taskCounter = getTaskCounter(taskCounterURL);
+
+ URL taskAttemptURL = getTaskAttemptURL(jobId, data.getTaskId(), data.getAttemptId());
+ long[] taskExecTime = getTaskExecTime(taskAttemptURL);
+
+ data.setCounter(taskCounter);
+ data.setTime(taskExecTime);
+ }
+ }
+
+ private String getTaskFailedStackTrace(URL taskAllAttemptsUrl) throws IOException, AuthenticationException {
+ JsonNode rootNode = ThreadContextMR2.readJsonNode(taskAllAttemptsUrl);
+ JsonNode tasks = rootNode.path("taskAttempts").path("taskAttempt");
+ for (JsonNode task : tasks) {
+ String state = task.get("state").getValueAsText();
+ if (!state.equals("FAILED")) {
+ continue;
+ }
+ String stacktrace = task.get("diagnostics").getValueAsText();
+ if (stacktrace.startsWith("Error:")) {
+ return stacktrace;
+ } else {
+ // This is not a valid stacktrace. Might due to a bug in hadoop2.3 and fixed in 2.4
+ throw new RuntimeException("This is not a valid stack trace.");
+ }
+ }
+ throw new RuntimeException("No failed task attempt in this failed task.");
+ }
+ }
+}
+
+final class ThreadContextMR2 {
+ private static final Logger logger = Logger.getLogger(ThreadContextMR2.class);
+ private static final AtomicInteger THREAD_ID = new AtomicInteger(1);
+
+ private static final ThreadLocal _LOCAL_THREAD_ID = new ThreadLocal() {
+ @Override
+ public Integer initialValue() {
+ return THREAD_ID.getAndIncrement();
+ }
+ };
+
+ private static final ThreadLocal _LOCAL_LAST_UPDATED = new ThreadLocal();
+ private static final ThreadLocal _LOCAL_UPDATE_INTERVAL = new ThreadLocal();
+
+ private static final ThreadLocal _LOCAL_DIAGNOSTIC_PATTERN = new ThreadLocal() {
+ @Override
+ public Pattern initialValue() {
+ // Example: "Task task_1443068695259_9143_m_000475 failed 1 times"
+ return Pattern.compile(
+ "Task[\\s\\u00A0]+(.*)[\\s\\u00A0]+failed[\\s\\u00A0]+([0-9])[\\s\\u00A0]+times[\\s\\u00A0]+");
+ }
+ };
+
+ private static final ThreadLocal _LOCAL_AUTH_TOKEN =
+ new ThreadLocal() {
+ @Override
+ public AuthenticatedURL.Token initialValue() {
+ _LOCAL_LAST_UPDATED.set(System.currentTimeMillis());
+ // Random an interval for each executor to avoid update token at the same time
+ _LOCAL_UPDATE_INTERVAL.set(Statistics.MINUTE_IN_MS * 30 + new Random().nextLong()
+ % (3 * Statistics.MINUTE_IN_MS));
+ logger.info("Executor " + _LOCAL_THREAD_ID.get() + " update interval " + _LOCAL_UPDATE_INTERVAL.get() * 1.0
+ / Statistics.MINUTE_IN_MS);
+ return new AuthenticatedURL.Token();
+ }
+ };
+
+ private static final ThreadLocal _LOCAL_AUTH_URL = new ThreadLocal() {
+ @Override
+ public AuthenticatedURL initialValue() {
+ return new AuthenticatedURL();
+ }
+ };
+
+ private static final ThreadLocal _LOCAL_MAPPER = new ThreadLocal() {
+ @Override
+ public ObjectMapper initialValue() {
+ return new ObjectMapper();
+ }
+ };
+
+ private ThreadContextMR2() {
+ // Empty on purpose
+ }
+
+ public static Matcher getDiagnosticMatcher(String diagnosticInfo) {
+ return _LOCAL_DIAGNOSTIC_PATTERN.get().matcher(diagnosticInfo);
+ }
+
+ public static JsonNode readJsonNode(URL url) throws IOException, AuthenticationException {
+ HttpURLConnection conn = _LOCAL_AUTH_URL.get().openConnection(url, _LOCAL_AUTH_TOKEN.get());
+ return _LOCAL_MAPPER.get().readTree(conn.getInputStream());
+ }
+
+ public static void updateAuthToken() {
+ long curTime = System.currentTimeMillis();
+ if (curTime - _LOCAL_LAST_UPDATED.get() > _LOCAL_UPDATE_INTERVAL.get()) {
+ logger.info("Executor " + _LOCAL_THREAD_ID.get() + " updates its AuthenticatedToken.");
+ _LOCAL_AUTH_TOKEN.set(new AuthenticatedURL.Token());
+ _LOCAL_AUTH_URL.set(new AuthenticatedURL());
+ _LOCAL_LAST_UPDATED.set(curTime);
+ }
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/ExceptionHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/ExceptionHeuristic.java
new file mode 100644
index 000000000..b5cb7e52f
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/ExceptionHeuristic.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+
+
+public class ExceptionHeuristic implements Heuristic {
+
+ private HeuristicConfigurationData _heuristicConfData;
+
+ public ExceptionHeuristic(HeuristicConfigurationData heuristicConfData) {
+ this._heuristicConfData = heuristicConfData;
+ }
+
+ @Override
+ public HeuristicConfigurationData getHeuristicConfData() {
+ return _heuristicConfData;
+ }
+
+ @Override
+ public HeuristicResult apply(MapReduceApplicationData data) {
+ if (data.getSucceeded()) {
+ return null;
+ }
+ HeuristicResult result = new HeuristicResult(
+ _heuristicConfData.getClassName(), _heuristicConfData.getHeuristicName(), Severity.MODERATE, 0);
+ String diagnosticInfo = data.getDiagnosticInfo();
+ if (diagnosticInfo != null) {
+ result.addResultDetail("Error", "Stacktrace", diagnosticInfo);
+ } else {
+ String msg = "Unable to find stacktrace info. Please find the real problem in the Jobhistory link above."
+ + "Exception can happen either in task log or Application Master log.";
+ result.addResultDetail("Error", msg);
+ }
+ return result;
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/GenericDataSkewHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/GenericDataSkewHeuristic.java
new file mode 100644
index 000000000..2c75e8bb0
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/GenericDataSkewHeuristic.java
@@ -0,0 +1,165 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.google.common.primitives.Longs;
+import com.linkedin.drelephant.analysis.HDFSContext;
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.math.Statistics;
+
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+import com.linkedin.drelephant.util.Utils;
+import java.util.Arrays;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.List;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.log4j.Logger;
+
+
+/**
+ * This Heuristic analyses the skewness in the task input data
+ */
+public abstract class GenericDataSkewHeuristic implements Heuristic {
+ private static final Logger logger = Logger.getLogger(GenericDataSkewHeuristic.class);
+
+ // Severity Parameters
+ private static final String NUM_TASKS_SEVERITY = "num_tasks_severity";
+ private static final String DEVIATION_SEVERITY = "deviation_severity";
+ private static final String FILES_SEVERITY = "files_severity";
+
+ // Default value of parameters
+ private double[] numTasksLimits = {10, 50, 100, 200}; // Number of map or reduce tasks
+ private double[] deviationLimits = {2, 4, 8, 16}; // Deviation in i/p bytes btw 2 groups
+ private double[] filesLimits = {1d/8, 1d/4, 1d/2, 1d}; // Fraction of HDFS Block Size
+
+ private MapReduceCounterData.CounterName _counterName;
+ private HeuristicConfigurationData _heuristicConfData;
+
+ private void loadParameters() {
+ Map paramMap = _heuristicConfData.getParamMap();
+ String heuristicName = _heuristicConfData.getHeuristicName();
+
+ double[] confNumTasksThreshold = Utils.getParam(paramMap.get(NUM_TASKS_SEVERITY), numTasksLimits.length);
+ if (confNumTasksThreshold != null) {
+ numTasksLimits = confNumTasksThreshold;
+ }
+ logger.info(heuristicName + " will use " + NUM_TASKS_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(numTasksLimits));
+
+ double[] confDeviationThreshold = Utils.getParam(paramMap.get(DEVIATION_SEVERITY), deviationLimits.length);
+ if (confDeviationThreshold != null) {
+ deviationLimits = confDeviationThreshold;
+ }
+ logger.info(heuristicName + " will use " + DEVIATION_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(deviationLimits));
+
+ double[] confFilesThreshold = Utils.getParam(paramMap.get(FILES_SEVERITY), filesLimits.length);
+ if (confFilesThreshold != null) {
+ filesLimits = confFilesThreshold;
+ }
+ logger.info(heuristicName + " will use " + FILES_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(filesLimits));
+ for (int i = 0; i < filesLimits.length; i++) {
+ filesLimits[i] = filesLimits[i] * HDFSContext.HDFS_BLOCK_SIZE;
+ }
+ }
+
+ protected GenericDataSkewHeuristic(MapReduceCounterData.CounterName counterName,
+ HeuristicConfigurationData heuristicConfData) {
+ this._counterName = counterName;
+ this._heuristicConfData = heuristicConfData;
+
+ loadParameters();
+ }
+
+ protected abstract MapReduceTaskData[] getTasks(MapReduceApplicationData data);
+
+ @Override
+ public HeuristicConfigurationData getHeuristicConfData() {
+ return _heuristicConfData;
+ }
+
+ @Override
+ public HeuristicResult apply(MapReduceApplicationData data) {
+
+ if(!data.getSucceeded()) {
+ return null;
+ }
+
+ MapReduceTaskData[] tasks = getTasks(data);
+
+ //Gather data
+ List inputBytes = new ArrayList();
+
+ for (int i = 0; i < tasks.length; i++) {
+ if (tasks[i].isSampled()) {
+ inputBytes.add(tasks[i].getCounters().get(_counterName));
+ }
+ }
+
+ // Ratio of total tasks / sampled tasks
+ double scale = ((double)tasks.length) / inputBytes.size();
+ //Analyze data. TODO: This is a temp fix. findTwogroups should support list as input
+ long[][] groups = Statistics.findTwoGroups(Longs.toArray(inputBytes));
+
+ long avg1 = Statistics.average(groups[0]);
+ long avg2 = Statistics.average(groups[1]);
+
+ long min = Math.min(avg1, avg2);
+ long diff = Math.abs(avg2 - avg1);
+
+ Severity severity = getDeviationSeverity(min, diff);
+
+ //This reduces severity if the largest file sizes are insignificant
+ severity = Severity.min(severity, getFilesSeverity(avg2));
+
+ //This reduces severity if number of tasks is insignificant
+ severity = Severity.min(severity, Severity.getSeverityAscending(
+ groups[0].length, numTasksLimits[0], numTasksLimits[1], numTasksLimits[2], numTasksLimits[3]));
+
+ HeuristicResult result = new HeuristicResult(_heuristicConfData.getClassName(),
+ _heuristicConfData.getHeuristicName(), severity, Utils.getHeuristicScore(severity, tasks.length));
+
+ result.addResultDetail("Number of tasks", Integer.toString(tasks.length));
+ result.addResultDetail("Group A", groups[0].length + " tasks @ " + FileUtils.byteCountToDisplaySize(avg1) + " avg");
+ result.addResultDetail("Group B", groups[1].length + " tasks @ " + FileUtils.byteCountToDisplaySize(avg2) + " avg");
+
+ return result;
+ }
+
+ private Severity getDeviationSeverity(long averageMin, long averageDiff) {
+ if (averageMin <= 0) {
+ averageMin = 1;
+ }
+ long value = averageDiff / averageMin;
+ return Severity.getSeverityAscending(
+ value, deviationLimits[0], deviationLimits[1], deviationLimits[2], deviationLimits[3]);
+ }
+
+ private Severity getFilesSeverity(long value) {
+ return Severity.getSeverityAscending(
+ value, filesLimits[0], filesLimits[1], filesLimits[2], filesLimits[3]);
+ }
+
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/GenericGCHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/GenericGCHeuristic.java
new file mode 100644
index 000000000..8513c0586
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/GenericGCHeuristic.java
@@ -0,0 +1,145 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+import com.linkedin.drelephant.util.Utils;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.math.Statistics;
+import java.util.Map;
+import org.apache.log4j.Logger;
+
+
+/**
+ * Analyses garbage collection efficiency
+ */
+public abstract class GenericGCHeuristic implements Heuristic {
+ private static final Logger logger = Logger.getLogger(GenericGCHeuristic.class);
+
+ // Severity Parameters
+ private static final String GC_RATIO_SEVERITY = "gc_ratio_severity";
+ private static final String RUNTIME_SEVERITY = "runtime_severity_in_min";
+
+ // Default value of parameters
+ private double[] gcRatioLimits = {0.01d, 0.02d, 0.03d, 0.04d}; // Garbage Collection Time / CPU Time
+ private double[] runtimeLimits = {5, 10, 12, 15}; // Task Runtime in milli sec
+
+ private HeuristicConfigurationData _heuristicConfData;
+
+ private void loadParameters() {
+ Map paramMap = _heuristicConfData.getParamMap();
+ String heuristicName = _heuristicConfData.getHeuristicName();
+
+ double[] confGcRatioThreshold = Utils.getParam(paramMap.get(GC_RATIO_SEVERITY), gcRatioLimits.length);
+ if (confGcRatioThreshold != null) {
+ gcRatioLimits = confGcRatioThreshold;
+ }
+ logger.info(heuristicName + " will use " + GC_RATIO_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(gcRatioLimits));
+
+ double[] confRuntimeThreshold = Utils.getParam(paramMap.get(RUNTIME_SEVERITY), runtimeLimits.length);
+ if (confRuntimeThreshold != null) {
+ runtimeLimits = confRuntimeThreshold;
+ }
+ logger.info(heuristicName + " will use " + RUNTIME_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(runtimeLimits));
+ for (int i = 0; i < runtimeLimits.length; i++) {
+ runtimeLimits[i] = runtimeLimits[i] * Statistics.MINUTE_IN_MS;
+ }
+ }
+
+ protected GenericGCHeuristic(HeuristicConfigurationData heuristicConfData) {
+ this._heuristicConfData = heuristicConfData;
+
+ loadParameters();
+ }
+
+ protected abstract MapReduceTaskData[] getTasks(MapReduceApplicationData data);
+
+ @Override
+ public HeuristicConfigurationData getHeuristicConfData() {
+ return _heuristicConfData;
+ }
+
+ @Override
+ public HeuristicResult apply(MapReduceApplicationData data) {
+
+ if(!data.getSucceeded()) {
+ return null;
+ }
+
+ MapReduceTaskData[] tasks = getTasks(data);
+ List gcMs = new ArrayList();
+ List cpuMs = new ArrayList();
+ List runtimesMs = new ArrayList();
+
+ for (MapReduceTaskData task : tasks) {
+ if (task.isSampled()) {
+ runtimesMs.add(task.getTotalRunTimeMs());
+ gcMs.add(task.getCounters().get(MapReduceCounterData.CounterName.GC_MILLISECONDS));
+ cpuMs.add(task.getCounters().get(MapReduceCounterData.CounterName.CPU_MILLISECONDS));
+ }
+ }
+
+ long avgRuntimeMs = Statistics.average(runtimesMs);
+ long avgCpuMs = Statistics.average(cpuMs);
+ long avgGcMs = Statistics.average(gcMs);
+ double ratio = avgCpuMs != 0 ? avgGcMs*(1.0)/avgCpuMs: 0;
+
+ Severity severity;
+ if (tasks.length == 0) {
+ severity = Severity.NONE;
+ } else {
+ severity = getGcRatioSeverity(avgRuntimeMs, avgCpuMs, avgGcMs);
+ }
+
+ HeuristicResult result = new HeuristicResult(_heuristicConfData.getClassName(),
+ _heuristicConfData.getHeuristicName(), severity, Utils.getHeuristicScore(severity, tasks.length));
+
+ result.addResultDetail("Number of tasks", Integer.toString(tasks.length));
+ result.addResultDetail("Avg task runtime (ms)", Long.toString(avgRuntimeMs));
+ result.addResultDetail("Avg task CPU time (ms)", Long.toString(avgCpuMs));
+ result.addResultDetail("Avg task GC time (ms)", Long.toString(avgGcMs));
+ result.addResultDetail("Task GC/CPU ratio", Double.toString(ratio));
+ return result;
+ }
+
+ private Severity getGcRatioSeverity(long runtimeMs, long cpuMs, long gcMs) {
+ double gcRatio = ((double)gcMs)/cpuMs;
+ Severity ratioSeverity = Severity.getSeverityAscending(
+ gcRatio, gcRatioLimits[0], gcRatioLimits[1], gcRatioLimits[2], gcRatioLimits[3]);
+
+ // Severity is reduced if task runtime is insignificant
+ Severity runtimeSeverity = getRuntimeSeverity(runtimeMs);
+
+ return Severity.min(ratioSeverity, runtimeSeverity);
+ }
+
+ private Severity getRuntimeSeverity(long runtimeMs) {
+ return Severity.getSeverityAscending(
+ runtimeMs, runtimeLimits[0], runtimeLimits[1], runtimeLimits[2], runtimeLimits[3]);
+ }
+
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/GenericMemoryHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/GenericMemoryHeuristic.java
new file mode 100644
index 000000000..03fd8c81e
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/GenericMemoryHeuristic.java
@@ -0,0 +1,192 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+import com.linkedin.drelephant.util.Utils;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.math.Statistics;
+
+import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import org.apache.log4j.Logger;
+
+
+/**
+ * This heuristic deals with the efficiency of container size
+ */
+public abstract class GenericMemoryHeuristic implements Heuristic {
+ private static final Logger logger = Logger.getLogger(GenericMemoryHeuristic.class);
+ private static final long CONTAINER_MEMORY_DEFAULT_BYTES = 2048L * FileUtils.ONE_MB;
+
+ // Severity Parameters
+ private static final String MEM_RATIO_SEVERITY = "memory_ratio_severity";
+ private static final String CONTAINER_MEM_SEVERITY = "container_memory_severity";
+ private static final String CONTAINER_MEM_DEFAULT_MB = "container_memory_default_mb";
+
+ // Default value of parameters
+ private double[] memRatioLimits = {0.6d, 0.5d, 0.4d, 0.3d}; // Avg Physical Mem of Tasks / Container Mem
+ private double[] memoryLimits = {1.1d, 1.5d, 2.0d, 2.5d}; // Container Memory Severity Limits
+
+ private String _containerMemConf;
+ private HeuristicConfigurationData _heuristicConfData;
+
+ private void loadParameters() {
+ Map paramMap = _heuristicConfData.getParamMap();
+ String heuristicName = _heuristicConfData.getHeuristicName();
+
+ double[] confMemRatioLimits = Utils.getParam(paramMap.get(MEM_RATIO_SEVERITY), memRatioLimits.length);
+ if (confMemRatioLimits != null) {
+ memRatioLimits = confMemRatioLimits;
+ }
+ logger.info(heuristicName + " will use " + MEM_RATIO_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(memRatioLimits));
+
+ long containerMemDefaultBytes = CONTAINER_MEMORY_DEFAULT_BYTES;
+ if (paramMap.containsKey(CONTAINER_MEM_DEFAULT_MB)) {
+ containerMemDefaultBytes = Long.valueOf(paramMap.get(CONTAINER_MEM_DEFAULT_MB)) * FileUtils.ONE_MB;
+ }
+ logger.info(heuristicName + " will use " + CONTAINER_MEM_DEFAULT_MB + " with the following threshold setting: "
+ + containerMemDefaultBytes);
+
+ double[] confMemoryLimits = Utils.getParam(paramMap.get(CONTAINER_MEM_SEVERITY), memoryLimits.length);
+ if (confMemoryLimits != null) {
+ memoryLimits = confMemoryLimits;
+ }
+ logger.info(heuristicName + " will use " + CONTAINER_MEM_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(memoryLimits));
+ for (int i = 0; i < memoryLimits.length; i++) {
+ memoryLimits[i] = memoryLimits[i] * containerMemDefaultBytes;
+ }
+ }
+
+ protected GenericMemoryHeuristic(String containerMemConf, HeuristicConfigurationData heuristicConfData) {
+ this._containerMemConf = containerMemConf;
+ this._heuristicConfData = heuristicConfData;
+
+ loadParameters();
+ }
+
+ protected abstract MapReduceTaskData[] getTasks(MapReduceApplicationData data);
+
+ @Override
+ public HeuristicConfigurationData getHeuristicConfData() {
+ return _heuristicConfData;
+ }
+
+ @Override
+ public HeuristicResult apply(MapReduceApplicationData data) {
+
+ if(!data.getSucceeded()) {
+ return null;
+ }
+
+ String containerSizeStr = data.getConf().getProperty(_containerMemConf);
+ if (containerSizeStr == null) {
+ return null;
+ }
+
+ long containerMem;
+ try {
+ containerMem = Long.parseLong(containerSizeStr);
+ } catch (NumberFormatException e) {
+ // Some job has a string var like "${VAR}" for this config.
+ if(containerSizeStr.startsWith("$")) {
+ String realContainerConf = containerSizeStr.substring(containerSizeStr.indexOf("{")+1,
+ containerSizeStr.indexOf("}"));
+ containerMem = Long.parseLong(data.getConf().getProperty(realContainerConf));
+ } else {
+ throw e;
+ }
+ }
+ containerMem *= FileUtils.ONE_MB;
+
+ MapReduceTaskData[] tasks = getTasks(data);
+ List taskPMems = new ArrayList();
+ List taskVMems = new ArrayList();
+ List runtimesMs = new ArrayList();
+ long taskPMin = Long.MAX_VALUE;
+ long taskPMax = 0;
+ for (MapReduceTaskData task : tasks) {
+ if (task.isSampled()) {
+ runtimesMs.add(task.getTotalRunTimeMs());
+ long taskPMem = task.getCounters().get(MapReduceCounterData.CounterName.PHYSICAL_MEMORY_BYTES);
+ long taskVMem = task.getCounters().get(MapReduceCounterData.CounterName.VIRTUAL_MEMORY_BYTES);
+ taskPMems.add(taskPMem);
+ taskPMin = Math.min(taskPMin, taskPMem);
+ taskPMax = Math.max(taskPMax, taskPMem);
+ taskVMems.add(taskVMem);
+ }
+ }
+
+ if(taskPMin == Long.MAX_VALUE) {
+ taskPMin = 0;
+ }
+
+ long taskPMemAvg = Statistics.average(taskPMems);
+ long taskVMemAvg = Statistics.average(taskVMems);
+ long averageTimeMs = Statistics.average(runtimesMs);
+
+ Severity severity;
+ if (tasks.length == 0) {
+ severity = Severity.NONE;
+ } else {
+ severity = getTaskMemoryUtilSeverity(taskPMemAvg, containerMem);
+ }
+
+ HeuristicResult result = new HeuristicResult(_heuristicConfData.getClassName(),
+ _heuristicConfData.getHeuristicName(), severity, Utils.getHeuristicScore(severity, tasks.length));
+
+ result.addResultDetail("Number of tasks", Integer.toString(tasks.length));
+ result.addResultDetail("Avg task runtime", Statistics.readableTimespan(averageTimeMs));
+ result.addResultDetail("Avg Physical Memory (MB)", Long.toString(taskPMemAvg / FileUtils.ONE_MB));
+ result.addResultDetail("Max Physical Memory (MB)", Long.toString(taskPMax / FileUtils.ONE_MB));
+ result.addResultDetail("Min Physical Memory (MB)", Long.toString(taskPMin / FileUtils.ONE_MB));
+ result.addResultDetail("Avg Virtual Memory (MB)", Long.toString(taskVMemAvg / FileUtils.ONE_MB));
+ result.addResultDetail("Requested Container Memory", FileUtils.byteCountToDisplaySize(containerMem));
+
+ return result;
+ }
+
+ private Severity getTaskMemoryUtilSeverity(long taskMemAvg, long taskMemMax) {
+ double ratio = ((double)taskMemAvg) / taskMemMax;
+ Severity sevRatio = getMemoryRatioSeverity(ratio);
+ // Severity is reduced if the requested container memory is close to default
+ Severity sevMax = getContainerMemorySeverity(taskMemMax);
+
+ return Severity.min(sevRatio, sevMax);
+ }
+
+
+ private Severity getContainerMemorySeverity(long taskMemMax) {
+ return Severity.getSeverityAscending(
+ taskMemMax, memoryLimits[0], memoryLimits[1], memoryLimits[2], memoryLimits[3]);
+ }
+
+ private Severity getMemoryRatioSeverity(double ratio) {
+ return Severity.getSeverityDescending(
+ ratio, memRatioLimits[0], memRatioLimits[1], memRatioLimits[2], memRatioLimits[3]);
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/JobQueueLimitHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/JobQueueLimitHeuristic.java
new file mode 100644
index 000000000..0ca8c0ce8
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/JobQueueLimitHeuristic.java
@@ -0,0 +1,120 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+
+
+public class JobQueueLimitHeuristic implements Heuristic {
+
+ private HeuristicConfigurationData _heuristicConfData;
+
+ protected JobQueueLimitHeuristic(HeuristicConfigurationData heuristicConfData) {
+ this._heuristicConfData = heuristicConfData;
+ }
+
+ @Override
+ public HeuristicConfigurationData getHeuristicConfData() {
+ return _heuristicConfData;
+ }
+
+ @Override
+ public HeuristicResult apply(MapReduceApplicationData data) {
+
+ HeuristicResult result = new HeuristicResult(_heuristicConfData.getClassName(),
+ _heuristicConfData.getHeuristicName(), Severity.NONE, 0);
+ Properties jobConf = data.getConf();
+ long queueTimeoutLimitMs = TimeUnit.MINUTES.toMillis(15);
+
+ // Fetch the Queue to which the job is submitted.
+ String queueName = jobConf.getProperty("mapred.job.queue.name");
+ if (queueName == null) {
+ throw new IllegalStateException("Queue Name not found.");
+ }
+
+ // Compute severity if job is submitted to default queue else set severity to NONE.
+ MapReduceTaskData[] mapTasks = data.getMapperData();
+ MapReduceTaskData[] redTasks = data.getReducerData();
+ Severity[] mapTasksSeverity = new Severity[mapTasks.length];
+ Severity[] redTasksSeverity = new Severity[redTasks.length];
+ if (queueName.equals("default")) {
+ result.addResultDetail("Queue: ", queueName, null);
+ result.addResultDetail("Number of Map tasks", Integer.toString(mapTasks.length));
+ result.addResultDetail("Number of Reduce tasks", Integer.toString(redTasks.length));
+
+ // Calculate Severity of Mappers
+ mapTasksSeverity = getTasksSeverity(mapTasks, queueTimeoutLimitMs);
+ result.addResultDetail("Number of Map tasks that are in severe state (14 to 14.5 min)",
+ Long.toString(getSeverityFrequency(Severity.SEVERE, mapTasksSeverity)));
+ result.addResultDetail("Number of Map tasks that are in critical state (over 14.5 min)",
+ Long.toString(getSeverityFrequency(Severity.CRITICAL, mapTasksSeverity)));
+
+ // Calculate Severity of Reducers
+ redTasksSeverity = getTasksSeverity(redTasks, queueTimeoutLimitMs);
+ result.addResultDetail("Number of Reduce tasks that are in severe state (14 to 14.5 min)",
+ Long.toString(getSeverityFrequency(Severity.SEVERE, redTasksSeverity)));
+ result.addResultDetail("Number of Reduce tasks that are in critical state (over 14.5 min)",
+ Long.toString(getSeverityFrequency(Severity.CRITICAL, redTasksSeverity)));
+
+ // Calculate Job severity
+ result.setSeverity(Severity.max(Severity.max(mapTasksSeverity), Severity.max(redTasksSeverity)));
+
+ } else {
+ result.addResultDetail("Not Applicable", "This Heuristic is not applicable to " + queueName + " queue");
+ result.setSeverity(Severity.NONE);
+ }
+ return result;
+ }
+
+ private Severity[] getTasksSeverity(MapReduceTaskData[] tasks, long queueTimeout) {
+ Severity[] tasksSeverity = new Severity[tasks.length];
+ int i = 0;
+ for (MapReduceTaskData task : tasks) {
+ tasksSeverity[i] = getQueueLimitSeverity(task.getTotalRunTimeMs(), queueTimeout);
+ i++;
+ }
+ return tasksSeverity;
+ }
+
+ private long getSeverityFrequency(Severity severity, Severity[] tasksSeverity) {
+ long count = 0;
+ for (Severity taskSeverity : tasksSeverity) {
+ if (taskSeverity.equals(severity)) {
+ count++;
+ }
+ }
+ return count;
+ }
+
+ private Severity getQueueLimitSeverity(long taskTime, long queueTimeout) {
+ long timeUnitMs = TimeUnit.SECONDS.toMillis(30); // 30s
+ if (queueTimeout == 0) {
+ return Severity.NONE;
+ }
+ return Severity.getSeverityAscending(taskTime, queueTimeout - 4 * timeUnitMs, queueTimeout - 3 * timeUnitMs,
+ queueTimeout - 2 * timeUnitMs, queueTimeout - timeUnitMs);
+ }
+
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/MapperDataSkewHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperDataSkewHeuristic.java
new file mode 100644
index 000000000..e598b6807
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperDataSkewHeuristic.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+
+
+/**
+ * This Heuristic analyses the skewness in the mapper input data
+ */
+public class MapperDataSkewHeuristic extends GenericDataSkewHeuristic {
+
+ public MapperDataSkewHeuristic(HeuristicConfigurationData heuristicConfData) {
+ super(MapReduceCounterData.CounterName.HDFS_BYTES_READ, heuristicConfData);
+ }
+
+ @Override
+ protected MapReduceTaskData[] getTasks(MapReduceApplicationData data) {
+ return data.getMapperData();
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/MapperGCHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperGCHeuristic.java
new file mode 100644
index 000000000..e57320128
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperGCHeuristic.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+
+
+public class MapperGCHeuristic extends GenericGCHeuristic {
+
+ public MapperGCHeuristic(HeuristicConfigurationData heuristicConfData) {
+ super(heuristicConfData);
+ }
+
+ @Override
+ protected MapReduceTaskData[] getTasks(MapReduceApplicationData data) {
+ return data.getMapperData();
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/MapperMemoryHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperMemoryHeuristic.java
new file mode 100644
index 000000000..b973b7450
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperMemoryHeuristic.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+
+
+public class MapperMemoryHeuristic extends GenericMemoryHeuristic {
+ public static final String MAPPER_MEMORY_CONF = "mapreduce.map.memory.mb";
+
+ public MapperMemoryHeuristic(HeuristicConfigurationData _heuristicConfData) {
+ super(MAPPER_MEMORY_CONF, _heuristicConfData);
+ }
+
+ @Override
+ protected MapReduceTaskData[] getTasks(MapReduceApplicationData data) {
+ return data.getMapperData();
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/MapperSpeedHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperSpeedHeuristic.java
new file mode 100644
index 000000000..fc719a5ba
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperSpeedHeuristic.java
@@ -0,0 +1,152 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+import com.linkedin.drelephant.util.Utils;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import com.linkedin.drelephant.analysis.HDFSContext;
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.math.Statistics;
+
+import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import org.apache.log4j.Logger;
+
+
+public class MapperSpeedHeuristic implements Heuristic {
+ private static final Logger logger = Logger.getLogger(MapperSpeedHeuristic.class);
+
+ // Severity parameters.
+ private static final String DISK_SPEED_SEVERITY = "disk_speed_severity";
+ private static final String RUNTIME_SEVERITY = "runtime_severity_in_min";
+
+ // Default value of parameters
+ private double[] diskSpeedLimits = {1d/2, 1d/4, 1d/8, 1d/32}; // Fraction of HDFS block size
+ private double[] runtimeLimits = {5, 10, 15, 30}; // The Map task runtime in milli sec
+
+ private HeuristicConfigurationData _heuristicConfData;
+
+ private void loadParameters() {
+ Map paramMap = _heuristicConfData.getParamMap();
+ String heuristicName = _heuristicConfData.getHeuristicName();
+
+ double[] confDiskSpeedThreshold = Utils.getParam(paramMap.get(DISK_SPEED_SEVERITY), diskSpeedLimits.length);
+ if (confDiskSpeedThreshold != null) {
+ diskSpeedLimits = confDiskSpeedThreshold;
+ }
+ logger.info(heuristicName + " will use " + DISK_SPEED_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(diskSpeedLimits));
+ for (int i = 0; i < diskSpeedLimits.length; i++) {
+ diskSpeedLimits[i] = diskSpeedLimits[i] * HDFSContext.DISK_READ_SPEED;
+ }
+
+ double[] confRuntimeThreshold = Utils.getParam(paramMap.get(RUNTIME_SEVERITY), runtimeLimits.length);
+ if (confRuntimeThreshold != null) {
+ runtimeLimits = confRuntimeThreshold;
+ }
+ logger.info(heuristicName + " will use " + RUNTIME_SEVERITY + " with the following threshold settings: " + Arrays
+ .toString(runtimeLimits));
+ for (int i = 0; i < runtimeLimits.length; i++) {
+ runtimeLimits[i] = runtimeLimits[i] * Statistics.MINUTE_IN_MS;
+ }
+ }
+
+ public MapperSpeedHeuristic(HeuristicConfigurationData heuristicConfData) {
+ this._heuristicConfData = heuristicConfData;
+ loadParameters();
+ }
+
+ @Override
+ public HeuristicConfigurationData getHeuristicConfData() {
+ return _heuristicConfData;
+ }
+
+ @Override
+ public HeuristicResult apply(MapReduceApplicationData data) {
+
+ if(!data.getSucceeded()) {
+ return null;
+ }
+
+ MapReduceTaskData[] tasks = data.getMapperData();
+
+ List inputByteSizes = new ArrayList();
+ List speeds = new ArrayList();
+ List runtimesMs = new ArrayList();
+
+ for (MapReduceTaskData task : tasks) {
+
+ if (task.isSampled()) {
+ long inputBytes = task.getCounters().get(MapReduceCounterData.CounterName.HDFS_BYTES_READ);
+ long runtimeMs = task.getTotalRunTimeMs();
+ inputByteSizes.add(inputBytes);
+ runtimesMs.add(runtimeMs);
+ //Speed is bytes per second
+ speeds.add((1000 * inputBytes) / (runtimeMs));
+ }
+ }
+
+ long medianSpeed;
+ long medianSize;
+ long medianRuntimeMs;
+
+ if (tasks.length != 0) {
+ medianSpeed = Statistics.median(speeds);
+ medianSize = Statistics.median(inputByteSizes);
+ medianRuntimeMs = Statistics.median(runtimesMs);
+ } else {
+ medianSpeed = 0;
+ medianSize = 0;
+ medianRuntimeMs = 0;
+ }
+
+ Severity severity = getDiskSpeedSeverity(medianSpeed);
+
+ //This reduces severity if task runtime is insignificant
+ severity = Severity.min(severity, getRuntimeSeverity(medianRuntimeMs));
+
+ HeuristicResult result = new HeuristicResult(_heuristicConfData.getClassName(),
+ _heuristicConfData.getHeuristicName(), severity, Utils.getHeuristicScore(severity, tasks.length));
+
+ result.addResultDetail("Number of tasks", Integer.toString(tasks.length));
+ result.addResultDetail("Median task input size", FileUtils.byteCountToDisplaySize(medianSize));
+ result.addResultDetail("Median task runtime", Statistics.readableTimespan(medianRuntimeMs));
+ result.addResultDetail("Median task speed", FileUtils.byteCountToDisplaySize(medianSpeed) + "/s");
+
+ return result;
+ }
+
+ private Severity getDiskSpeedSeverity(long speed) {
+ return Severity.getSeverityDescending(
+ speed, diskSpeedLimits[0], diskSpeedLimits[1], diskSpeedLimits[2], diskSpeedLimits[3]);
+ }
+
+ private Severity getRuntimeSeverity(long runtimeMs) {
+ return Severity.getSeverityAscending(
+ runtimeMs, runtimeLimits[0], runtimeLimits[1], runtimeLimits[2], runtimeLimits[3]);
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/MapperSpillHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperSpillHeuristic.java
new file mode 100644
index 000000000..54c875734
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperSpillHeuristic.java
@@ -0,0 +1,140 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+import com.linkedin.drelephant.util.Utils;
+import java.util.Arrays;
+import java.util.Map;
+import org.apache.log4j.Logger;
+
+
+public class MapperSpillHeuristic implements Heuristic {
+ private static final Logger logger = Logger.getLogger(MapperSpillHeuristic.class);
+ private static final long THRESHOLD_SPILL_FACTOR = 10000;
+
+ // Severity parameters.
+ private static final String SPILL_SEVERITY = "spill_severity";
+ private static final String NUM_TASKS_SEVERITY = "num_tasks_severity";
+
+ // Default value of parameters
+ private double[] numTasksLimits = {50, 100, 500, 1000}; // Number of Map tasks.
+ private double[] spillLimits = {2.01d, 2.2d, 2.5d, 3.0d}; // Records spilled/total output records
+
+ private HeuristicConfigurationData _heuristicConfData;
+
+ private void loadParameters() {
+ Map paramMap = _heuristicConfData.getParamMap();
+ String heuristicName = _heuristicConfData.getHeuristicName();
+
+ double[] confNumTasksThreshold = Utils.getParam(paramMap.get(NUM_TASKS_SEVERITY), numTasksLimits.length);
+ if (confNumTasksThreshold != null) {
+ numTasksLimits = confNumTasksThreshold;
+ }
+ logger.info(heuristicName + " will use " + NUM_TASKS_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(numTasksLimits));
+
+ double[] confSpillThreshold = Utils.getParam(paramMap.get(SPILL_SEVERITY), spillLimits.length);
+ if (confSpillThreshold != null) {
+ spillLimits = confSpillThreshold;
+ }
+ logger.info(heuristicName + " will use " + SPILL_SEVERITY + " with the following threshold settings: " + Arrays
+ .toString(spillLimits));
+ for (int i = 0; i < spillLimits.length; i++) {
+ spillLimits[i] = spillLimits[i] * THRESHOLD_SPILL_FACTOR;
+ }
+ }
+
+ public MapperSpillHeuristic(HeuristicConfigurationData heuristicConfData) {
+ this._heuristicConfData = heuristicConfData;
+ loadParameters();
+ }
+
+ @Override
+ public HeuristicConfigurationData getHeuristicConfData() {
+ return _heuristicConfData;
+ }
+
+ @Override
+ public HeuristicResult apply(MapReduceApplicationData data) {
+
+ if(!data.getSucceeded()) {
+ return null;
+ }
+
+ MapReduceTaskData[] tasks = data.getMapperData();
+
+ long totalSpills = 0;
+ long totalOutputRecords = 0;
+ double ratioSpills = 0.0;
+
+ for (MapReduceTaskData task : tasks) {
+
+ if (task.isSampled()) {
+ totalSpills += task.getCounters().get(MapReduceCounterData.CounterName.SPILLED_RECORDS);
+ totalOutputRecords += task.getCounters().get(MapReduceCounterData.CounterName.MAP_OUTPUT_RECORDS);
+ }
+ }
+
+ //If both totalSpills and totalOutputRecords are zero then set ratioSpills to zero.
+ if (totalSpills == 0) {
+ ratioSpills = 0;
+ } else {
+ ratioSpills = (double) totalSpills / (double) totalOutputRecords;
+ }
+
+ Severity severity = getSpillSeverity(ratioSpills);
+
+ // Severity is reduced if number of tasks is small
+ Severity taskSeverity = getNumTasksSeverity(tasks.length);
+ severity = Severity.min(severity, taskSeverity);
+
+ HeuristicResult result = new HeuristicResult(_heuristicConfData.getClassName(),
+ _heuristicConfData.getHeuristicName(), severity, Utils.getHeuristicScore(severity, tasks.length));
+
+ result.addResultDetail("Number of tasks", Integer.toString(tasks.length));
+ result.addResultDetail("Avg spilled records per task",
+ tasks.length == 0 ? "0" : Long.toString(totalSpills / tasks.length));
+ result.addResultDetail("Avg output records per task",
+ tasks.length == 0 ? "0" : Long.toString(totalOutputRecords / tasks.length));
+ result.addResultDetail("Ratio of spilled records to output records", Double.toString(ratioSpills));
+
+ return result;
+
+ }
+
+ private Severity getSpillSeverity(double ratioSpills) {
+
+ long normalizedSpillRatio = 0;
+ //Normalize the ratio to integer.
+ normalizedSpillRatio = (long) (ratioSpills * THRESHOLD_SPILL_FACTOR);
+
+ return Severity.getSeverityAscending(
+ normalizedSpillRatio, spillLimits[0], spillLimits[1], spillLimits[2], spillLimits[3]);
+ }
+
+ private Severity getNumTasksSeverity(long numTasks) {
+ return Severity.getSeverityAscending(
+ numTasks, numTasksLimits[0], numTasksLimits[1], numTasksLimits[2], numTasksLimits[3]);
+ }
+}
diff --git a/app/com/linkedin/drelephant/mapreduce/heuristics/MapperTimeHeuristic.java b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperTimeHeuristic.java
new file mode 100644
index 000000000..2b7f32564
--- /dev/null
+++ b/app/com/linkedin/drelephant/mapreduce/heuristics/MapperTimeHeuristic.java
@@ -0,0 +1,170 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed 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 com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+import com.linkedin.drelephant.configurations.heuristic.HeuristicConfigurationData;
+import com.linkedin.drelephant.util.Utils;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.mapreduce.data.MapReduceTaskData;
+import com.linkedin.drelephant.math.Statistics;
+
+import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import org.apache.log4j.Logger;
+
+
+public class MapperTimeHeuristic implements Heuristic {
+ private static final Logger logger = Logger.getLogger(MapperTimeHeuristic.class);
+
+ // Severity parameters.
+ private static final String SHORT_RUNTIME_SEVERITY = "short_runtime_severity_in_min";
+ private static final String LONG_RUNTIME_SEVERITY = "long_runtime_severity_in_min";
+ private static final String NUM_TASKS_SEVERITY = "num_tasks_severity";
+
+ // Default value of parameters
+ private double[] shortRuntimeLimits = {10, 4, 2, 1}; // Limits(ms) for tasks with shorter runtime
+ private double[] longRuntimeLimits = {15, 30, 60, 120}; // Limits(ms) for tasks with longer runtime
+ private double[] numTasksLimits = {50, 101, 500, 1000}; // Number of Map tasks.
+
+ private HeuristicConfigurationData _heuristicConfData;
+
+ private void loadParameters() {
+ Map paramMap = _heuristicConfData.getParamMap();
+ String heuristicName = _heuristicConfData.getHeuristicName();
+
+ double[] confShortThreshold = Utils.getParam(paramMap.get(SHORT_RUNTIME_SEVERITY), shortRuntimeLimits.length);
+ if (confShortThreshold != null) {
+ shortRuntimeLimits = confShortThreshold;
+ }
+ logger.info(heuristicName + " will use " + SHORT_RUNTIME_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(shortRuntimeLimits));
+ for (int i = 0; i < shortRuntimeLimits.length; i++) {
+ shortRuntimeLimits[i] = shortRuntimeLimits[i] * Statistics.MINUTE_IN_MS;
+ }
+
+ double[] confLongThreshold = Utils.getParam(paramMap.get(LONG_RUNTIME_SEVERITY), longRuntimeLimits.length);
+ if (confLongThreshold != null) {
+ longRuntimeLimits = confLongThreshold;
+ }
+ logger.info(heuristicName + " will use " + LONG_RUNTIME_SEVERITY + " with the following threshold settings: "
+ + Arrays.toString(longRuntimeLimits));
+ for (int i = 0; i < longRuntimeLimits.length; i++) {
+ longRuntimeLimits[i] = longRuntimeLimits[i] * Statistics.MINUTE_IN_MS;
+ }
+
+ double[] confNumTasksThreshold = Utils.getParam(paramMap.get(NUM_TASKS_SEVERITY), numTasksLimits.length);
+ if (confNumTasksThreshold != null) {
+ numTasksLimits = confNumTasksThreshold;
+ }
+ logger.info(heuristicName + " will use " + NUM_TASKS_SEVERITY + " with the following threshold settings: " + Arrays
+ .toString(numTasksLimits));
+ }
+
+ public MapperTimeHeuristic(HeuristicConfigurationData heuristicConfData) {
+ this._heuristicConfData = heuristicConfData;
+ loadParameters();
+ }
+
+ @Override
+ public HeuristicConfigurationData getHeuristicConfData() {
+ return _heuristicConfData;
+ }
+
+ @Override
+ public HeuristicResult apply(MapReduceApplicationData data) {
+
+ if(!data.getSucceeded()) {
+ return null;
+ }
+
+ MapReduceTaskData[] tasks = data.getMapperData();
+
+ List