Partitioner

// ItemPartition.java: orders and revenue per item with 4 reducers.
// -D part.mode=hash      default HashPartitioner
// -D part.mode=counter   CounterPartitioner: every item of one canteen counter (A–D) goes to the same reducer
// -D part.mode=bad       BadPartitioner (BadPartitioner.java): returns an invalid partition number
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
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 ItemPartition {
  public static class M extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final Text item = new Text();
    private final IntWritable amount = new IntWritable();
    @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
      item.set(f[3]); amount.set(Integer.parseInt(f[4]));
      ctx.write(item, amount);
    }
  }

  // Counter A: chai, coffee   B: samosa, vada pav   C: poha, idli   D: dosa, thali
  public static class CounterPartitioner extends Partitioner<Text, IntWritable> {
    private static final Map<String, Integer> COUNTER = new HashMap<>();
    static {
      COUNTER.put("chai", 0); COUNTER.put("coffee", 0); COUNTER.put("samosa", 1); COUNTER.put("vada pav", 1);
      COUNTER.put("poha", 2); COUNTER.put("idli", 2); COUNTER.put("dosa", 3); COUNTER.put("thali", 3);
    }
    @Override
    public int getPartition(Text item, IntWritable amount, int numReduceTasks) {
      Integer c = COUNTER.get(item.toString());
      if (c == null) return 0;                           // unknown items go to the first reducer
      return c % numReduceTasks;                         // must always be in 0 .. numReduceTasks-1
    }
  }

  public static class R extends Reducer<Text, IntWritable, Text, Text> {
    @Override
    protected void reduce(Text item, Iterable<IntWritable> amounts, Context ctx) throws IOException, InterruptedException {
      long n = 0, sum = 0;
      for (IntWritable a : amounts) { n++; sum += a.get(); }
      ctx.write(item, new Text(n + "\t" + sum));
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    String mode = conf.get("part.mode", "hash");
    Job job = Job.getInstance(conf, "item partition " + mode);
    job.setJarByClass(ItemPartition.class);
    job.setMapperClass(M.class);
    job.setReducerClass(R.class);
    if (mode.equals("counter")) job.setPartitionerClass(CounterPartitioner.class);
    if (mode.equals("bad")) job.setPartitionerClass(BadPartitioner.class);
    job.setNumReduceTasks(conf.getInt("part.reducers", 4));           // -D part.reducers=N to change
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(IntWritable.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 ItemPartition.java lab8-namenode-1:/tmp/mr9/
//   docker cp BadPartitioner.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 .

// BadPartitioner.java: a partitioner that returns a number outside 0..numReduceTasks-1 for one item
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class BadPartitioner extends Partitioner<Text, IntWritable> {
  @Override
  public int getPartition(Text item, IntWritable amount, int numReduceTasks) {
    return item.toString().length();                     // "vada pav" -> 8, but there are only 4 reducers
  }
}

# reduce_tasks.sh JOB_ID : for each reduce task, where it ran, how long it took and how many records it got
J=$1; ID=${J#job_}; H=http://historyserver:19888/ws/v1/history/mapreduce/jobs/$J/tasks
for t in $(curl -s $H | grep -o "task_${ID}_r_[0-9]*" | sort -u); do
  curl -s $H/$t/counters | python3 -c '
import sys, json
t = sys.argv[1]
groups = json.load(sys.stdin)["jobTaskCounters"]["taskCounterGroup"]
groups = groups if isinstance(groups, list) else [groups]
c = {}
for g in groups:
    cs = g["counter"] if isinstance(g["counter"], list) else [g["counter"]]
    c.update({x["name"]: x["value"] for x in cs})
print("%s  reduce input records %9d  groups %d" % (t[-8:], c["REDUCE_INPUT_RECORDS"], c["REDUCE_INPUT_GROUPS"]))' $t
done

# Inside the namenode container:
#   bash reduce_tasks.sh job_<your job id>

# partitioner_lab.sh: who decides which reducer gets which key? Default hash vs a custom partitioner.
cd /tmp/mr9
D=/user/hadoop/data/student_orders.csv
for mode in hash counter; do
  hdfs dfs -rm -r -f -skipTrash /out/part_$mode > /dev/null 2>&1
  hadoop jar mr9.jar ItemPartition -D part.mode=$mode $D /out/part_$mode > /tmp/part_$mode.log 2>&1
  echo "== part.mode=$mode, 4 reducers"
  for f in $(hdfs dfs -ls /out/part_$mode | grep part-r | awk '{print $8}'); do
    echo "$(basename $f): $(hdfs dfs -cat $f | awk -F'\t' '{printf "%s(%s) ", $1, $2}')"
  done
  sleep 3
  bash reduce_tasks.sh $(grep -o "job_[0-9_]*" /tmp/part_$mode.log | head -1)
done
echo "== a partitioner that returns 8 for 4 reducers"
hdfs dfs -rm -r -f -skipTrash /out/part_bad > /dev/null 2>&1
hadoop jar mr9.jar ItemPartition -D part.mode=bad $D /out/part_bad 2>&1 | grep -E "Illegal partition|Task Id|failed with state|Job .* failed|Failed map tasks" | sed -E 's/^[0-9-]+ [0-9:]+ INFO  //' | cut -c1-160 | sort -u | head -6

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