@Override
  public HeuristicResult apply(MapReduceApplicationData data) {

    if(!data.getSucceeded()) {
      return null;
    }

    String containerSizeStr = data.getConf().getProperty(_containerMemConf);
    long containerMem = -1L;

    if (containerSizeStr != null) {
      try {
        containerMem = Long.parseLong(containerSizeStr);
      } catch (NumberFormatException e0) {
        // Some job has a string var like "${VAR}" for this config.
        if(containerSizeStr.startsWith("$")) {
          String realContainerConf = containerSizeStr.substring(containerSizeStr.indexOf("{")+1,
              containerSizeStr.indexOf("}"));
          String realContainerSizeStr = data.getConf().getProperty(realContainerConf);
          try {
            containerMem = Long.parseLong(realContainerSizeStr);
          }
          catch (NumberFormatException e1) {
            logger.warn(realContainerConf + ": expected number [" + realContainerSizeStr + "]");
          }
        } else {
          logger.warn(_containerMemConf + ": expected number [" + containerSizeStr + "]");
        }
      }
    }
    if (containerMem < 0) {
      containerMem = getContainerMemDefaultMBytes();
    }
    containerMem *= FileUtils.ONE_MB;

    MapReduceTaskData[] tasks = getTasks(data);
    List<Long> taskPMems = new ArrayList<Long>();
    List<Long> taskVMems = new ArrayList<Long>();
    List<Long> runtimesMs = new ArrayList<Long>();
    long taskPMin = Long.MAX_VALUE;
    long taskPMax = 0;
    for (MapReduceTaskData task : tasks) {
      if (task.isTimeAndCounterDataPresent()) {
        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;
  }
This heuristic deals with the efficiency of container size
app/com/linkedin/drelephant/mapreduce/heuristics/GenericMemoryHeuristic.java