Counters

// OrderCounters.java: a map-only job that only counts. Enum counters, dynamic counters,
// and reading every counter back in the driver after the job.
// -D counters.perStudent=true adds one dynamic counter per student (to hit the counter limit)
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.CounterGroup;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.TaskCounter;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class OrderCounters {
  enum Category { DRINK, SNACK, BREAKFAST, MEAL, UNKNOWN }

  public static class M extends Mapper<LongWritable, Text, NullWritable, NullWritable> {
    private boolean perStudent;
    @Override
    protected void setup(Context ctx) { perStudent = ctx.getConfiguration().getBoolean("counters.perStudent", false); }

    @Override
    protected void map(LongWritable k, Text line, Context ctx) {
      String[] f = line.toString().split(",");          // order_id,student_id,time,item,amount
      String item = f[3];
      Category c;
      switch (item) {
        case "chai": case "coffee": c = Category.DRINK; break;
        case "samosa": case "vada pav": c = Category.SNACK; break;
        case "poha": case "idli": c = Category.BREAKFAST; break;
        case "dosa": case "thali": c = Category.MEAL; break;
        default: c = Category.UNKNOWN;
      }
      ctx.getCounter(c).increment(1);                             // enum counter
      ctx.getCounter("Orders by hour", f[2].substring(11, 13)).increment(1);   // dynamic counter: group, name
      ctx.getCounter("Revenue (Rs)", c.name()).increment(Long.parseLong(f[4]));
      if (perStudent) ctx.getCounter("Orders by student", f[1]).increment(1);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    Job job = Job.getInstance(conf, "order counters");
    job.setJarByClass(OrderCounters.class);
    job.setMapperClass(M.class);
    job.setNumReduceTasks(0);
    job.setOutputFormatClass(NullOutputFormat.class);            // no output files at all
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    boolean ok = job.waitForCompletion(false);                   // false: don't print progress and counters

    // The driver reads the final counters and decides what to do with them
    Counters cs = job.getCounters();
    long in = cs.findCounter(TaskCounter.MAP_INPUT_RECORDS).getValue();
    System.out.println("job " + (ok ? "SUCCEEDED" : "FAILED") + ", map input records = " + in);
    for (String group : new String[] {OrderCounters.Category.class.getName(), "Orders by hour", "Revenue (Rs)"}) {
      CounterGroup g = cs.getGroup(group);
      System.out.println("[" + g.getDisplayName() + "]");
      for (Counter c : g) System.out.printf("  %-10s %,d%n", c.getDisplayName(), c.getValue());
    }
    if (cs.findCounter(Category.UNKNOWN).getValue() > 0) { System.out.println("unknown items found!"); System.exit(2); }
    System.exit(ok ? 0 : 1);
  }
}

// In the mr-lab folder:
//   python3 make_student_orders.py
//   docker exec lab8-namenode-1 mkdir -p /tmp/mr9
//   docker cp OrderCounters.java lab8-namenode-1:/tmp/mr9/
//   docker cp student_orders.csv lab8-namenode-1:/tmp/mr9/
//   docker exec -it lab8-namenode-1 bash
// Inside the container:
//   cd /tmp/mr9
//   hdfs dfs -D dfs.blocksize=8388608 -put student_orders.csv /user/hadoop/data/
//   mkdir -p classes && javac -cp $(hadoop classpath) -d classes *.java && jar cf mr9.jar -C classes .
//   hadoop jar mr9.jar OrderCounters /user/hadoop/data/student_orders.csv

# counters_lab.sh: enum and dynamic counters read back by the driver; then too many counters
cd /tmp/mr9
D=/user/hadoop/data/student_orders.csv
echo "== OrderCounters (map-only, NullOutputFormat: no output files)"
hadoop jar mr9.jar OrderCounters $D > /tmp/oc.log 2>&1; echo "exit code: $?"
grep -vE "INFO|WARN" /tmp/oc.log
echo "== one dynamic counter per student (2,000 of them)"
hadoop jar mr9.jar OrderCounters -D counters.perStudent=true $D > /tmp/oc2.log 2>&1; echo "exit code: $?"
grep "job FAILED" /tmp/oc2.log
app=$(grep -o "application_[0-9_]*" /tmp/oc2.log | head -1)
yarn application -status $app 2>/dev/null | grep -A1 "Diagnostics :" | sed 's/^\s*//'
echo "-- why the map attempts failed (JobHistory REST)"
J=job_${app#application_}
curl -s http://historyserver:19888/ws/v1/history/mapreduce/jobs/$J/tasks/${J/job/task}_m_000000/attempts \
  | python3 -c 'import sys, json; t = json.load(sys.stdin)["taskAttempts"]["taskAttempt"]; t = t if isinstance(t, list) else [t]; print(next(a for a in t if a["state"] == "FAILED")["diagnostics"].splitlines()[0])'
echo "== the limit"
hdfs getconf -confKey mapreduce.job.counters.max

# Inside the namenode container, in /tmp/mr9 (attempts are read from the JobHistory Server REST API):
#   bash counters_lab.sh