package com.manning.hip.ch6;


import com.manning.hip.common.PaddedTable;
import org.apache.commons.lang.reflect.MethodUtils;
import org.apache.hadoop.mapred.JobHistory;

import java.lang.reflect.InvocationTargetException;
import java.text.ParseException;
import java.util.List;

public final class MetricSummary {

  public static void main(String... args) throws Exception {
    try {
      dumpTaskTimes(args);
    } catch (Throwable t) {
      t.printStackTrace();
    }
  }

  public static void dumpTaskTimes(String... args)
      throws Exception {
    printTasks(JobHistoryHelper.getJobInfoFromCliArgs(args));
  }

  public static void printTasks(JobHistory.JobInfo job)
      throws ParseException, IllegalAccessException,
      InvocationTargetException, NoSuchMethodException {

    List<TaskMetrics> mapMetrics = JobHistoryHelper.getMapTaskMetrics(
        job);
    List<TaskMetrics> reduceMetrics =
        JobHistoryHelper.getReduceTaskMetrics(
            job);

    System.out.println("Job Statistics");
    System.out.println("");

    PaddedTable table = new PaddedTable();
    table
        .addColumnTitle("Item")
        .addColumnTitle("Min")
        .addColumnTitle("Max")
        .addColumnTitle("Median")
        .addColumnTitle("Mean")
        .addColumnTitle("StdDev");

    dumpTasks(table, mapMetrics, "execution time",
        "getOverallTimeMillis", true);
    dumpTasks(table, mapMetrics, "input records", "getInputRecords",
        false);
    dumpTasks(table, mapMetrics, "input bytes", "getInputBytes", false);
    dumpTasks(table, mapMetrics, "output records", "getOutputRecords",
        false);
    dumpTasks(table, mapMetrics, "output bytes", "getOutputBytes",
        false);

    DataSkewMetrics.decorateHeader("MAP TASKS");
    System.out.println("");
    System.out.println("Num Map Tasks:    " + mapMetrics.size());
    System.out.println("");
    System.out.println(table);
    table.clearRows();

    dumpTasks(table, reduceMetrics, "execution time",
        "getOverallTimeMillis", true);
    dumpTasks(table, reduceMetrics, "shuffle time",
        "getShuffleTimeMillis", true);
    dumpTasks(table, reduceMetrics, "sort time", "getSortTimeMillis",
        true);
    dumpTasks(table, reduceMetrics, "input records", "getInputRecords",
        false);
    dumpTasks(table, reduceMetrics, "input bytes", "getInputBytes",
        false);
    dumpTasks(table, reduceMetrics, "output records",
        "getOutputRecords", false);
    dumpTasks(table, reduceMetrics, "output bytes", "getOutputBytes",
        false);

    DataSkewMetrics.decorateHeader("REDUCE TASKS");
    System.out.println("");
    System.out.println("Num Reduce Tasks: " + reduceMetrics.size());
    System.out.println("");
    System.out.println(table);
  }

  public static void dumpTasks(PaddedTable table,
                               List<TaskMetrics> metrics,
                               String heading,
                               String fieldName,
                               boolean isTime)
      throws IllegalAccessException, InvocationTargetException,
      NoSuchMethodException {

    table.newRow();
    table.addColumnValue(heading);

    if (metrics.size() == 0) {
      table.addColumnValue("")
          .addColumnValue("")
          .addColumnValue("")
          .addColumnValue("");
    }

    Long minVal = null;
    Long maxVal = null;
    long totalVals = 0;

    long vals[] = new long[metrics.size()];
    int i = 0;

    for (TaskMetrics m : metrics) {
      long v = extractLongFieldValue(m, fieldName);
      minVal = minVal == null ? v : Math.min(minVal, v);
      maxVal = maxVal == null ? v : Math.max(maxVal, v);
      totalVals += v;
      vals[i++] = v;
    }

    double mean = totalVals / metrics.size();
    double stddev = stddev(vals);
    double median = median(vals);

    if (isTime) {
      table.addColumnValue(JobHistoryHelper.formatTime(minVal))
          .addColumnValue(JobHistoryHelper.formatTime(maxVal))
          .addColumnValue(JobHistoryHelper.formatTime((long) median))
          .addColumnValue(JobHistoryHelper.formatTime((long) mean))
          .addColumnValue(JobHistoryHelper.formatTime((long) stddev));
    } else {
      table.addColumnValue(minVal)
          .addColumnValue(maxVal)
          .addColumnValue((long) median)
          .addColumnValue((long) mean)
          .addColumnValue((long) stddev);
    }
  }

  public static double stddev(long[] m) {
    long total = 0;
    for(long l: m) {
      total += l;
    }
    double mean = total / m.length;

    long diffMean = 0;
    for(long l: m) {
      diffMean += Math.abs(mean - l);
    }

    return Math.sqrt((double) diffMean / (double) m.length);
  }

  public static double median(long[] m) {
    int middle = m.length / 2;
    if (m.length % 2 == 1) {
      return m[middle];
    } else {
      return (m[middle - 1] + m[middle]) / 2.0;
    }
  }

  public static long extractLongFieldValue(TaskMetrics m,
                                           String fieldName)
      throws IllegalAccessException, InvocationTargetException,
      NoSuchMethodException {
    return (Long) MethodUtils.invokeMethod(m, fieldName, null);
  }
}

@@ -1933,24 +1933,62 @@
 earRows();%0A%0A
+    if(reduceMetrics.size() %3E 0) %7B%0A%0A  
     dumpTask
@@ -2021,32 +2021,34 @@
 xecution time%22,%0A
+  
         %22getOver
@@ -2066,32 +2066,34 @@
 is%22, true);%0A    
+  
 dumpTasks(table,
@@ -2128,24 +2128,26 @@
 e%22,%0A        
+  
 %22getShuffleT
@@ -2161,32 +2161,34 @@
 is%22, true);%0A    
+ 
+ 
 dumpTasks(table,
@@ -2245,23 +2245,27 @@
         
+  
 true);%0A
+  
     dump
@@ -2332,32 +2332,34 @@
 s%22,%0A        
+  
 false);%0A
     dumpTask
@@ -2338,32 +2338,34 @@
         false);%0A
+  
     dumpTasks(ta
@@ -2420,32 +2420,34 @@
 s%22,%0A        
+ 
+ 
 false);%0A
     dumpTask
@@ -2426,32 +2426,34 @@
         false);%0A
+  
     dumpTasks(ta
@@ -2482,32 +2482,34 @@
 utput records%22,%0A
+  
         %22getOutp
@@ -2524,32 +2524,34 @@
 s%22, false);%0A    
+  
 dumpTasks(table,
@@ -2600,32 +2600,34 @@
 Bytes%22,%0A        
+  
 false);%0A%0A    Dat
@@ -2615,32 +2615,34 @@
    false);%0A%0A    
+ 
+ 
 DataSkewMetrics.
@@ -2669,32 +2669,34 @@
 CE TASKS%22);%0A    
+  
 System.out.print
@@ -2695,32 +2695,34 @@
 ut.println(%22%22);%0A
+  
     System.out.p
@@ -2770,32 +2770,34 @@
 cs.size());%0A    
+  
 System.out.print
@@ -2796,32 +2796,34 @@
 ut.println(%22%22);%0A
+  
     System.out.p
@@ -2837,16 +2837,22 @@
 table);%0A
+    %7D%0A
   %7D%0A%0A  p

handle map-only jobs