Reduce-side join

// JoinReduceSide.java: reduce-side join of orders with the menu.
// Both inputs are mapped to (item, tagged record); the reducer meets them per item and writes
// every order with its category added.
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
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.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class JoinReduceSide {
  enum Join { ORDERS_JOINED, ORDERS_WITHOUT_MENU, MENU_FIRST, MENU_NOT_FIRST }

  // orders: order_id,student_id,time,item,amount   ->  (item, "O|whole line")
  public static class OrderMapper extends Mapper<LongWritable, Text, Text, Text> {
    private final Text item = new Text(), val = new Text();
    @Override
    protected void map(LongWritable k, Text line, Context ctx) throws IOException, InterruptedException {
      item.set(line.toString().split(",")[3]);
      val.set("O|" + line);
      ctx.write(item, val);
    }
  }

  // menu: item,category,price,counter   ->  (item, "M|category")
  public static class MenuMapper extends Mapper<LongWritable, Text, Text, Text> {
    private final Text item = new Text(), val = new Text();
    @Override
    protected void map(LongWritable k, Text line, Context ctx) throws IOException, InterruptedException {
      String[] f = line.toString().split(",");
      item.set(f[0]); val.set("M|" + f[1]);
      ctx.write(item, val);
    }
  }

  // For one item: one menu record and thousands of order records, in NO particular order
  public static class JoinReducer extends Reducer<Text, Text, Text, NullWritable> {
    private final Text out = new Text();
    @Override
    protected void reduce(Text item, Iterable<Text> vals, Context ctx) throws IOException, InterruptedException {
      String category = null;
      List<String> orders = new ArrayList<>();         // must buffer: the menu record may arrive last
      boolean first = true;
      for (Text v : vals) {
        String s = v.toString();
        if (s.startsWith("M|")) {
          category = s.substring(2);
          ctx.getCounter(first ? Join.MENU_FIRST : Join.MENU_NOT_FIRST).increment(1);
        } else orders.add(s.substring(2));
        first = false;
      }
      if (category == null) { ctx.getCounter(Join.ORDERS_WITHOUT_MENU).increment(orders.size()); return; }
      for (String o : orders) { out.set(o + "," + category); ctx.write(out, NullWritable.get()); }
      ctx.getCounter(Join.ORDERS_JOINED).increment(orders.size());
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();   // orders menu out
    Job job = Job.getInstance(conf, "reduce-side join");
    job.setJarByClass(JoinReduceSide.class);
    MultipleInputs.addInputPath(job, new Path(rest[0]), TextInputFormat.class, OrderMapper.class);
    MultipleInputs.addInputPath(job, new Path(rest[1]), TextInputFormat.class, MenuMapper.class);
    job.setReducerClass(JoinReducer.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(NullWritable.class);
    FileOutputFormat.setOutputPath(job, new Path(rest[2]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

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

# join_lab.sh: join 5 lakh orders with the 8-line menu, reduce-side and map-side, and compare
cd /tmp/mr9
O=/user/hadoop/data/student_orders.csv; MENU=/user/hadoop/data/menu.csv
hdfs dfs -put -f menu.csv $MENU
echo "== menu.csv (minus 'coffee', to show an order without a menu row)"
grep -v coffee menu.csv > menu_nocoffee.csv; hdfs dfs -put -f menu_nocoffee.csv /user/hadoop/data/
cat menu_nocoffee.csv
for j in JoinReduceSide JoinMapSide; do
  hdfs dfs -rm -r -f -skipTrash /out/$j > /dev/null 2>&1
  start=$(date +%s%N)
  hadoop jar mr9.jar $j $O /user/hadoop/data/menu_nocoffee.csv /out/$j > /tmp/$j.log 2>&1
  echo "== $j   $(( ($(date +%s%N) - start) / 1000000 )) ms"
  grep -E "number of splits|Launched (map|reduce) tasks|Map input records|Map output records|Reduce shuffle bytes|Reduce input groups|ORDERS_JOINED|ORDERS_WITHOUT_MENU|MENU_(NOT_)?FIRST|Bytes Written" /tmp/$j.log | sed -E 's/^[0-9-]+ [0-9:]+ INFO  //; s/^\s*/  /'
  hdfs dfs -ls /out/$j | grep part | awk '{print "  " $5, $8}'
done
echo "== a joined row"
hdfs dfs -cat /out/JoinMapSide/part-m-00000 2>/dev/null | head -2
echo "== same rows from both joins?"
diff <(hdfs dfs -cat /out/JoinReduceSide/part-* | sort) <(hdfs dfs -cat /out/JoinMapSide/part-* | sort) > /dev/null && echo "identical (459,636 rows each, compared after sorting)"
echo "== map task log of JoinMapSide: the cached menu"
app=$(grep -o "application_[0-9_]*" /tmp/JoinMapSide.log | head -1); sleep 5
yarn logs -applicationId $app -log_files stderr 2>/dev/null | grep -E "cache files|menu loaded" | head -2 | cut -c1-200

# Inside the namenode container:  bash /tmp/mr9/join_lab.sh