public HeuristicResult apply(TezApplicationData data) {

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

    TezTaskData[] tasks = data.getScopeTasks();

    List<Long> inputSizes = new ArrayList<Long>();
    List<Long> speeds = new ArrayList<Long>();
    List<Long> runtimesMs = new ArrayList<Long>();

    for (TezTaskData task : tasks) {

      if (task.isSampled()) {

        long inputBytes = 0;

        for (TezCounterData.CounterName counterName: _counterNames) {
          inputBytes += task.getCounters().get(counterName);
        }

        long runtimeMs = task.getTotalRunTimeMs();
        inputSizes.add(inputBytes);
        runtimesMs.add(runtimeMs);
        //Speed is records per second
        speeds.add((1000 * inputBytes) / (runtimeMs));
      }
    }

    long medianSpeed;
    long medianSize;
    long medianRuntimeMs;

    if (tasks.length != 0) {
      medianSpeed = Statistics.median(speeds);
      medianSize = Statistics.median(inputSizes);
      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 ", FileUtils.byteCountToDisplaySize(medianSize));
    result.addResultDetail("Median task runtime", Statistics.readableTimespan(medianRuntimeMs));
    result.addResultDetail("Median task speed", FileUtils.byteCountToDisplaySize(medianSpeed) + "/s");

    return result;
  }
Analyzes mapper task speed and efficiency
app/com/linkedin/drelephant/tez/heuristics/TezScopeSpeedHeuristic.java