Secondary sort

// FirstLastOrder.java: secondary sort. For every student: first order, last order and the
// longest gap between two orders, reading the orders in time order WITHOUT sorting in the reducer.
// Key = (student, time): sorted by both, partitioned and grouped by student only.
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class FirstLastOrder {

  public static class StudentTime implements WritableComparable<StudentTime> {
    String student = "", time = "";
    public StudentTime() { }
    @Override public void write(DataOutput out) throws IOException { out.writeUTF(student); out.writeUTF(time); }
    @Override public void readFields(DataInput in) throws IOException { student = in.readUTF(); time = in.readUTF(); }
    @Override public int compareTo(StudentTime o) {               // the SORT order: student, then time
      int c = student.compareTo(o.student);
      return c != 0 ? c : time.compareTo(o.time);
    }
    @Override public int hashCode() { return student.hashCode(); }
    @Override public boolean equals(Object o) { return o instanceof StudentTime && compareTo((StudentTime) o) == 0; }
  }

  // 1. Partition by student only, so all of a student's orders reach the same reducer
  public static class StudentPartitioner extends Partitioner<StudentTime, Text> {
    @Override public int getPartition(StudentTime k, Text v, int n) { return (k.student.hashCode() & Integer.MAX_VALUE) % n; }
  }

  // 2. Group by student only, so one reduce() call gets ALL of a student's orders (already time-sorted)
  public static class StudentGrouping extends WritableComparator {
    public StudentGrouping() { super(StudentTime.class, true); }
    @Override public int compare(WritableComparable a, WritableComparable b) {
      return ((StudentTime) a).student.compareTo(((StudentTime) b).student);
    }
  }

  public static class M extends Mapper<LongWritable, Text, StudentTime, Text> {
    private final StudentTime key = new StudentTime();
    private final Text val = new Text();
    @Override
    protected void map(LongWritable k, Text line, Context ctx) throws IOException, InterruptedException {
      String[] f = line.toString().split(",");          // order_id,student_id,time,item,amount
      key.student = f[1]; key.time = f[2];
      val.set(f[2] + " " + f[3]);
      ctx.write(key, val);
    }
  }

  public static class R extends Reducer<StudentTime, Text, Text, Text> {
    private static final DateTimeFormatter F = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm");
    private boolean sorted = true;
    @Override
    protected void reduce(StudentTime k, Iterable<Text> vals, Context ctx) throws IOException, InterruptedException {
      String first = null, last = null, prev = null, gapFrom = null, gapTo = null;
      long n = 0, maxGap = -1;
      for (Text v : vals) {                              // arrives in time order thanks to the sort
        String s = v.toString();
        if (first == null) first = s;
        if (prev != null) {
          if (s.substring(0, 16).compareTo(prev.substring(0, 16)) < 0) sorted = false;   // compare the time part only
          long gap = Duration.between(LocalDateTime.parse(prev.substring(0, 16), F), LocalDateTime.parse(s.substring(0, 16), F)).toMinutes();
          if (gap > maxGap) { maxGap = gap; gapFrom = prev.substring(0, 16); gapTo = s.substring(0, 16); }
        }
        prev = s; last = s; n++;
      }
      ctx.write(new Text(k.student), new Text(n + " orders | first " + first + " | last " + last
          + " | longest gap " + (maxGap / 60) + " h (" + gapFrom + " -> " + gapTo + ")"));
    }
    @Override
    protected void cleanup(Context ctx) { System.err.println("values were in time order in every group: " + sorted); }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    Job job = Job.getInstance(conf, "first last order (secondary sort)");
    job.setJarByClass(FirstLastOrder.class);
    job.setMapperClass(M.class);
    job.setReducerClass(R.class);
    job.setPartitionerClass(StudentPartitioner.class);
    if (conf.getBoolean("ss.grouping", true)) job.setGroupingComparatorClass(StudentGrouping.class);
    job.setMapOutputKeyClass(StudentTime.class);
    job.setMapOutputValueClass(Text.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    FileOutputFormat.setOutputPath(job, new Path(rest[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

// In the mr-lab folder:
//   docker exec lab8-namenode-1 mkdir -p /tmp/mr9
//   docker cp FirstLastOrder.java lab8-namenode-1:/tmp/mr9/
//   docker exec -it lab8-namenode-1 bash
// Inside the container:
//   cd /tmp/mr9
//   mkdir -p classes && javac -cp $(hadoop classpath) -d classes *.java && jar cf mr9.jar -C classes .
//   hadoop jar mr9.jar FirstLastOrder /user/hadoop/data/student_orders.csv /out/ss_true

# secondary_sort_lab.sh: values in time order inside each student's group, with and without a grouping comparator
cd /tmp/mr9
D=/user/hadoop/data/student_orders.csv
for g in true false; do
  hdfs dfs -rm -r -f -skipTrash /out/ss_$g > /dev/null 2>&1
  start=$(date +%s%N)
  hadoop jar mr9.jar FirstLastOrder -D ss.grouping=$g $D /out/ss_$g > /tmp/ss_$g.log 2>&1
  echo "== grouping comparator: $g   $(( ($(date +%s%N) - start) / 1000000 )) ms"
  grep -E "Reduce input records|Reduce input groups|Reduce output records" /tmp/ss_$g.log | sed 's/^\s*/  /'
  hdfs dfs -cat /out/ss_$g/part-r-00000 2>/dev/null | head -3 | cut -c1-200
done
echo "== a light user and the heaviest user"
hdfs dfs -cat /out/ss_true/part-r-00000 | grep -E "^S1413|^S0001" | cut -c1-200
echo "== the reducer's own check (task log)"
app=$(grep -o "application_[0-9_]*" /tmp/ss_true.log | head -1); sleep 5
yarn logs -applicationId $app -log_files stderr 2>/dev/null | grep "time order"
echo "== check in plain Linux: S1413's first and last order"
grep ",S1413," student_orders.csv | cut -d, -f3,4 | sort | sed -n '1p;$p'

# Inside the namenode container, in /tmp/mr9 (student_orders.csv copied there too):
#   bash secondary_sort_lab.sh