Check for tez and hive configs for memory container size in Tez heuristics

app-conf/HeuristicConf.xml

@@ -245,6 +245,19 @@
     </params>-->
   </heuristic>
 
+  <heuristic>
+    <applicationtype>mapreduce</applicationtype>
+    <heuristicname>Mapper Data Locality</heuristicname>
+    <classname>com.linkedin.drelephant.mapreduce.heuristics.MapperTaskLocalityHeuristic</classname>
+    <viewname>views.html.help.mapreduce.helpMapperTaskLocality$$$</viewname>
+    <!--<params>
+      <num_tasks_severity>10, 50, 100, 200</num_tasks_severity>
+      <deviation_severity>2, 4, 8, 16</deviation_severity>
+      <files_severity>1/8, 1/4, 1/2, 1</files_severity>
+    </params>-->
+  </heuristic>
+
+
   <heuristic>
     <applicationtype>mapreduce</applicationtype>
     <heuristicname>Reducer Skew</heuristicname>

app-conf/JobTypeConf.xml

@@ -43,17 +43,11 @@
     <conf>pig.script</conf>
   </jobType>
   <jobType>
-    <name>Hive-Tez</name>
-    <applicationtype>TEZ</applicationtype>
-        <conf>hive.execution.engine</conf>
+    <name>Tez</name>
+    <applicationtype>tez</applicationtype>
+    <conf>hive.mapred.mode</conf>
     <isDefault/>
   </jobType>
-  <jobType>
-  <name>Pig-Tez</name>
-   <applicationtype>TEZ</applicationtype>
-        <conf>exectype</conf>
-        <value>TEZ</value>
-  </jobType>
   <jobType>
     <name>Hive</name>
     <applicationtype>mapreduce</applicationtype>
@@ -86,4 +80,4 @@
     <conf>mapred.child.java.opts</conf>
     <isDefault/>
   </jobType>
-</jobTypes>
+</jobTypes>

app/com/linkedin/drelephant/mapreduce/data/MapReduceCounterData.java

@@ -98,6 +98,7 @@ public static enum GroupName {
     FileInput,
     FileSystemCounters,
     MapReduce,
+    JobCounters,
     FileOutput;
   }
 
@@ -131,12 +132,19 @@ public static enum CounterName {
     COMBINE_OUTPUT_RECORDS(GroupName.MapReduce, "COMBINE_OUTPUT_RECORDS", "Combine output records"),
     SPILLED_RECORDS(GroupName.MapReduce, "SPILLED_RECORDS", "Spilled Records"),
 
+    DATA_LOCAL_MAPS(GroupName.JobCounters,"DATA_LOCAL_MAPS","Data Local Map tasks"),
+    RACK_LOCAL_MAPS(GroupName.JobCounters,"RACK_LOCAL_MAPS","Rack Local Map Tasks"),
+    OTHER_LOCAL_MAPS(GroupName.JobCounters,"OTHER_LOCAL_MAPS","Other Local Maps"),
+    TOTAL_LAUNCHED_MAPS(GroupName.JobCounters,"TOTAL_LAUNCHED_MAPS","Total Launched Map Tasks"),
+
     CPU_MILLISECONDS(GroupName.MapReduce, "CPU_MILLISECONDS", "CPU time spent (ms)"),
     GC_MILLISECONDS(GroupName.MapReduce, "GC_TIME_MILLIS", "GC time elapsed (ms)"),
     COMMITTED_HEAP_BYTES(GroupName.MapReduce, "COMMITTED_HEAP_BYTES", "Total committed heap usage (bytes)"),
     PHYSICAL_MEMORY_BYTES(GroupName.MapReduce, "PHYSICAL_MEMORY_BYTES", "Physical memory (bytes) snapshot"),
     VIRTUAL_MEMORY_BYTES(GroupName.MapReduce, "VIRTUAL_MEMORY_BYTES", "Virtual memory (bytes) snapshot");
 
+
+
     GroupName _group;
     String _name;
     String _displayName;

app/com/linkedin/drelephant/mapreduce/heuristics/MapperTaskLocalityHeuristic.java

@@ -0,0 +1,107 @@
+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.configurations.heuristic.HeuristicConfigurationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.data.MapReduceCounterData;
+import com.linkedin.drelephant.util.Utils;
+import org.apache.log4j.Logger;
+
+import java.util.Arrays;
+import java.util.Map;
+
+public class MapperTaskLocalityHeuristic implements Heuristic<MapReduceApplicationData> {
+
+    private static final Logger logger = Logger.getLogger(MapperTaskLocalityHeuristic.class);
+
+    // Severity parameters.
+    private static final String NODE_LOCALITY_RATIO = "node_locality_severity";
+    private static final String RACK_LOCALITY_RATIO = "rack_locality_severity";
+    private static final String COUNTER_GROUP_NAME = "org.apache.hadoop.mapreduce.JobCounter";
+    private static final String DATA_LOCAL_MAPS = "DATA_LOCAL_MAPS";
+    private static final String RACK_LOCAL_MAPS = "RACK_LOCAL_MAPS";
+    private static final String OTHER_LOCAL_MAPS = "OTHER_LOCAL_MAPS";
+    private static final String TOTAL_LAUNCHED_MAPS = "TOTAL_LAUNCHED_MAPS";
+
+    private double[] nodeLocalityRatioLimits = {0.5d, 0.4d, 0.3d, 0.2d};
+    private double[] rackLocalityRatioLimits = {0.7d, 0.6d, 0.5d, 0.4d};
+
+    private HeuristicConfigurationData _heuristicConfData;
+
+    private void loadParameters() {
+        Map<String, String> paramMap = _heuristicConfData.getParamMap();
+        String heuristicName = _heuristicConfData.getHeuristicName();
+
+        double[] nodeLocalityThreshold = Utils.getParam(paramMap.get(NODE_LOCALITY_RATIO), nodeLocalityRatioLimits.length);
+        if (nodeLocalityThreshold != null) {
+            nodeLocalityRatioLimits = nodeLocalityThreshold;
+        }
+        logger.info(heuristicName + " will use " + NODE_LOCALITY_RATIO + " with the following threshold settings: "
+                + Arrays.toString(nodeLocalityRatioLimits));
+
+        double[] rackLocalityThreshold = Utils.getParam(paramMap.get(RACK_LOCALITY_RATIO), rackLocalityRatioLimits.length);
+        if (nodeLocalityThreshold != null) {
+            rackLocalityRatioLimits = rackLocalityThreshold;
+        }
+        logger.info(heuristicName + " will use " + RACK_LOCALITY_RATIO + " with the following threshold settings: "
+                + Arrays.toString(nodeLocalityRatioLimits));
+
+    }
+
+    public MapperTaskLocalityHeuristic(HeuristicConfigurationData heuristicConfData) {
+        this._heuristicConfData = heuristicConfData;
+        loadParameters();
+    }
+
+    @Override
+    public HeuristicConfigurationData getHeuristicConfData() {
+        return _heuristicConfData;
+    }
+
+    @Override
+    public HeuristicResult apply(MapReduceApplicationData data) {
+
+        if(!data.getSucceeded()) {
+            return null;
+        }
+
+        MapReduceCounterData currentJobData = data.getCounters();
+
+        Map<String,Long> totalCounters = currentJobData.getAllCountersInGroup(COUNTER_GROUP_NAME);
+
+        long dataLocalMaps = totalCounters.containsKey(DATA_LOCAL_MAPS) ? totalCounters.get(DATA_LOCAL_MAPS) : 0;
+        long rackLocalMaps = totalCounters.containsKey(RACK_LOCAL_MAPS) ? totalCounters.get(RACK_LOCAL_MAPS) : 0;
+        long otherLocalMaps = totalCounters.containsKey(OTHER_LOCAL_MAPS) ? totalCounters.get(OTHER_LOCAL_MAPS) : 0;
+        long totalLaunchedMaps = totalCounters.get(TOTAL_LAUNCHED_MAPS);
+
+        long nodeLocalityRatio = (dataLocalMaps + otherLocalMaps) / totalLaunchedMaps;
+        long rackLocalityRatio = rackLocalMaps/totalLaunchedMaps;
+
+
+        Severity severity = getTaskLocalitySeverity(nodeLocalityRatio);
+
+        severity = Severity.min(severity, getTaskLocalitySeverity(rackLocalityRatio));
+
+        HeuristicResult result = new HeuristicResult(_heuristicConfData.getClassName(),
+                _heuristicConfData.getHeuristicName(), severity, Utils.getHeuristicScore(severity, data.getMapperData().length));
+
+        result.addResultDetail("Data Local Map Tasks", Long.toString(dataLocalMaps));
+        result.addResultDetail("Other Local Map Tasks", Long.toString(otherLocalMaps));
+        result.addResultDetail("Rack Local Map Tasks", Long.toString(rackLocalMaps));
+        result.addResultDetail("Total Launched Map Tasks", Long.toString(totalLaunchedMaps));
+
+        return result;
+    }
+
+    private Severity getTaskLocalitySeverity(double ratio) {
+        return Severity.getSeverityDescending(
+                ratio, nodeLocalityRatioLimits[0], nodeLocalityRatioLimits[1], nodeLocalityRatioLimits[2], nodeLocalityRatioLimits[3]);
+    }
+
+
+
+
+}

app/com/linkedin/drelephant/tez/TezMetricsAggregator.java

@@ -15,6 +15,8 @@
  */
 package com.linkedin.drelephant.tez;
 
+
+import com.google.common.base.Strings;
 import com.linkedin.drelephant.analysis.*;
 import com.linkedin.drelephant.configurations.aggregator.AggregatorConfigurationData;
 import com.linkedin.drelephant.tez.data.TezApplicationData;
@@ -29,7 +31,9 @@ public class TezMetricsAggregator implements HadoopMetricsAggregator {
 
   private static final Logger logger = Logger.getLogger(TezMetricsAggregator.class);
 
-  private static final String TEZ_CONTAINER_CONFIG = "hive.tez.container.size";
+
+  private static final String TEZ_CONTAINER_CONFIG = "tez.task.resource.memory.mb";
+  private static final String HIVE_TEZ_CONTAINER_CONFIG = "hive.tez.container.size";
   private static final String MAP_CONTAINER_CONFIG = "mapreduce.map.memory.mb";
   private static final String REDUCER_CONTAINER_CONFIG = "mapreduce.reduce.memory.mb";
   private static final String REDUCER_SLOW_START_CONFIG = "mapreduce.job.reduce.slowstart.completedmaps";
@@ -84,26 +88,37 @@ public HadoopAggregatedData getResult() {
   }
 
   private long getMapContainerSize(HadoopApplicationData data) {
-    try {
-      long mapContainerSize = Long.parseLong(data.getConf().getProperty(TEZ_CONTAINER_CONFIG));
-      if (mapContainerSize > 0)
-        return mapContainerSize;
-      else
-        return Long.parseLong(data.getConf().getProperty(MAP_CONTAINER_CONFIG));
-    } catch ( NumberFormatException ex) {
-      return CONTAINER_MEMORY_DEFAULT_BYTES;
+    long mapperContainerSize;
+    if(!Strings.isNullOrEmpty(data.getConf().getProperty(TEZ_CONTAINER_CONFIG)) && Long.valueOf(data.getConf().getProperty(TEZ_CONTAINER_CONFIG)) > 0){
+      mapperContainerSize = Long.valueOf(data.getConf().getProperty(TEZ_CONTAINER_CONFIG));
+    }
+    else if(!Strings.isNullOrEmpty(data.getConf().getProperty(HIVE_TEZ_CONTAINER_CONFIG)) && Long.valueOf(data.getConf().getProperty(HIVE_TEZ_CONTAINER_CONFIG)) > 0){
+      mapperContainerSize = Long.valueOf(data.getConf().getProperty(HIVE_TEZ_CONTAINER_CONFIG));
+    }
+    else if(!Strings.isNullOrEmpty(data.getConf().getProperty(MAP_CONTAINER_CONFIG)) && Long.valueOf(data.getConf().getProperty(MAP_CONTAINER_CONFIG)) > 0){
+      mapperContainerSize = Long.valueOf(data.getConf().getProperty(MAP_CONTAINER_CONFIG));
+    }
+    else {
+      mapperContainerSize = CONTAINER_MEMORY_DEFAULT_BYTES;
     }
+    return mapperContainerSize;
   }
 
   private long getReducerContainerSize(HadoopApplicationData data) {
-    try {
-      long reducerContainerSize = Long.parseLong(data.getConf().getProperty(TEZ_CONTAINER_CONFIG));
-      if (reducerContainerSize > 0)
-        return reducerContainerSize;
-      else
-        return Long.parseLong(data.getConf().getProperty(REDUCER_CONTAINER_CONFIG));
-    } catch ( NumberFormatException ex) {
-      return CONTAINER_MEMORY_DEFAULT_BYTES;
+    long reducerContainerSize;
+    if(!Strings.isNullOrEmpty(data.getConf().getProperty(TEZ_CONTAINER_CONFIG)) && Long.valueOf(data.getConf().getProperty(TEZ_CONTAINER_CONFIG)) > 0){
+      reducerContainerSize = Long.valueOf(data.getConf().getProperty(TEZ_CONTAINER_CONFIG));
     }
+    else if(!Strings.isNullOrEmpty(data.getConf().getProperty(HIVE_TEZ_CONTAINER_CONFIG)) && Long.valueOf(data.getConf().getProperty(HIVE_TEZ_CONTAINER_CONFIG)) > 0){
+      reducerContainerSize = Long.valueOf(data.getConf().getProperty(HIVE_TEZ_CONTAINER_CONFIG));
+    }
+    else if(!Strings.isNullOrEmpty(data.getConf().getProperty(REDUCER_CONTAINER_CONFIG)) && Long.valueOf(data.getConf().getProperty(REDUCER_CONTAINER_CONFIG)) > 0){
+      reducerContainerSize = Long.valueOf(data.getConf().getProperty(REDUCER_CONTAINER_CONFIG));
+    }
+    else {
+      reducerContainerSize = CONTAINER_MEMORY_DEFAULT_BYTES;
+    }
+    return reducerContainerSize;
+
   }
 }

app/com/linkedin/drelephant/tez/heuristics/GenericMemoryHeuristic.java

@@ -82,6 +82,7 @@ private void loadParameters() {
   }
 
 
+
   public GenericMemoryHeuristic(String mapredContainerMemConf, HeuristicConfigurationData heuristicConfData) {
     this._mapredContainerMemConf = mapredContainerMemConf;
     this._heuristicConfData = heuristicConfData;
@@ -135,7 +136,6 @@ public HeuristicResult apply(TezApplicationData data) {
 
     String containerSizeStr;
 
-
     if(!Strings.isNullOrEmpty(data.getConf().getProperty(TEZ_MAPPER_MEMORY_CONF)) && Long.valueOf(data.getConf().getProperty(TEZ_MAPPER_MEMORY_CONF)) > 0){
       containerSizeStr = data.getConf().getProperty(TEZ_MAPPER_MEMORY_CONF);
     }
@@ -144,6 +144,7 @@ else if(!Strings.isNullOrEmpty(data.getConf().getProperty(HIVE_MAPPER_MEMORY_CON
     }
     else if(!Strings.isNullOrEmpty(data.getConf().getProperty(_mapredContainerMemConf)) && Long.valueOf(data.getConf().getProperty(_mapredContainerMemConf)) > 0) {
       containerSizeStr = data.getConf().getProperty(_mapredContainerMemConf);
+
     }
     else {
       containerSizeStr = getContainerMemDefaultMBytes();

app/com/linkedin/drelephant/tez/heuristics/MapperMemoryHeuristic.java

@@ -33,8 +33,10 @@ public class MapperMemoryHeuristic extends GenericMemoryHeuristic {
   private static final Logger logger = Logger.getLogger(MapperMemoryHeuristic.class);
   public static final String MAPRED_MAPPER_MEMORY_CONF = "mapreduce.map.memory.mb";
 
+
   public MapperMemoryHeuristic(HeuristicConfigurationData __heuristicConfData) {
     super(MAPRED_MAPPER_MEMORY_CONF, __heuristicConfData);
+
   }
 
   @Override

app/com/linkedin/drelephant/tez/heuristics/ReducerMemoryHeuristic.java

@@ -34,6 +34,7 @@ public class ReducerMemoryHeuristic extends GenericMemoryHeuristic {
 
   public ReducerMemoryHeuristic(HeuristicConfigurationData __heuristicConfData) {
     super(MAPRED_REDUCER_MEMORY_CONF, __heuristicConfData);
+
   }
 
   @Override

app/com/linkedin/drelephant/util/InfoExtractor.java

@@ -22,7 +22,6 @@
 import com.linkedin.drelephant.configurations.scheduler.SchedulerConfigurationData;
 
 import com.linkedin.drelephant.tez.data.TezApplicationData;
-import com.linkedin.drelephant.clients.WorkflowClient;
 
 import com.linkedin.drelephant.mapreduce.data.MapReduceApplicationData;
 import com.linkedin.drelephant.schedulers.Scheduler;

app/views/help/mapreduce/helpMapperTaskLocality.scala.html

@@ -0,0 +1,71 @@
+@*
+* 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.
+*@
+
+<p>
+    This analysis shows the data locality for your apps. <br>
+    We check the ratio between your job's node local map tasks AND the total number of map tasks. <br>
+    Also check the ratio between your job's rack local map tasks AND the total number of tasks <br>
+    The setting in question here is container memory is "yarn.scheduler.capacity.node-locality-delay". This is the Number of missed scheduling opportunities after which the capacity-scheduler attempts to schedule
+     rack local containers. <br>
+     For Cloud based storage systems: This number should be low (0 or -1) as there is no notion of data locality and there is no point even checking for it<br>
+     For on-prem storage/hdfs with cloud or on-prem compute, this number should be equal to the number of nodes in the cluster to guarantee a scheduling attempt to each data local node.
+    If this heuristic is above MODERATE, it means you are making good use of hadoop's most important feature of data locality.
+</p>
+
+<h5>Example</h5>
+<p>
+<div class="list-group">
+    <a class="list-group-item list-group-item-danger" href="#">
+        <h4 class="list-group-item-heading">Mapper Task Locality</h4>
+        <table class="list-group-item-text table table-condensed left-table">
+            <thead><tr><th colspan="2">Severity: Critical</th></tr></thead>
+            <tbody>
+            <tr>
+                <td>Data Local Map Tasks</td>
+                <td>1</td>
+            </tr>
+            <tr>
+                <td>Other Local Map Tasks</td>
+                <td>1</td>
+            </tr>
+            <tr>
+                <td>Rack local Map Tasks</td>
+                <td>56</td>
+            </tr>
+            <tr>
+                <td>Total Launched Map Tasks</td>
+                <td>127</td>
+            </tr>
+            </tbody>
+        </table>
+    </a>
+</div>
+</p>
+<h4>Suggestions</h4>
+<p>
+
+   <br>
+    The problem indicates your tasks are not running local to the data. This will cause large network data flow and could lead to bottlenecks for very large jobs.
+   <br>
+    You should try to increase <strong>yarn.scheduler.capacity.node-locality-delay</strong> to atleast the number of nodes on the cluster
+   <br>
+   See yarn.scheduler.fair.locality.threshold.node and yarn.scheduler.fair.locality.threshold.rack for fair scheduler
+    This will make sure that there is at least one scheduling attempt to each data local node
+    <br>
+
+    See <a href="https://hadoop.apache.org/docs/r2.7.3/hadoop-yarn/hadoop-yarn-site/CapacityScheduler.html">Capacity Scheduler Node Locality Settings</a> for further information.<br>
+    <br>
+    See <a href="https://hadoop.apache.org/docs/r2.7.3/hadoop-yarn/hadoop-yarn-site/CapacityScheduler.html">Fair Scheduler Node Locality Settings</a> for further information.<br>