@Override
  public HeuristicResult apply(TezApplicationData data) {
    if(!data.getSucceeded()) {
      return null;
    }
    TezTaskData[] tasks = data.getMapTaskData();

    long totalSpills = 0;
    long totalOutputRecords = 0;
    double ratioSpills = 0.0;

    for (TezTaskData task : tasks) {

      if (task.isSampled()) {
        totalSpills += task.getCounters().get(TezCounterData.CounterName.SPILLED_RECORDS);
        totalOutputRecords += task.getCounters().get(TezCounterData.CounterName.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;

  }
Analyzes mapper task data spill rates
app/com/linkedin/drelephant/tez/heuristics/MapperSpillHeuristic.java