Hadoop for Beginners (14): How MapReduce Shuffle Works

Tutorial index: Big Data for Beginners: tutorial series

The previous tutorial introduced how MapReduce executes and sketched out InputFormat. Now it’s time to talk about Shuffle. The data processing that happens after the Map method and before the Reduce method is called Shuffle.

MapReduce is the heart of Hadoop, and Shuffle is the heart of MapReduce — a lot of magic happens here. Since this is a beginner tutorial I’ll only give you a rough idea, enough to get you through the door; for the deep parts, consult a search engine and other sources.

Partitioning

MapReduce provides the Partitioner interface, whose job is to decide which ReduceTask should eventually handle the current output pair, based on the key or value and the number of reducers. By default it hashes the key and takes the modulus over the number of ReduceTasks. That default modulus only aims to balance load across reducers; if you have your own requirements, you can customize a Partitioner and set it on the job.

Writing a Custom Partitioner

To customize partitioning, just extend Partitioner and do your thing. Here’s a simple example that partitions to different ReduceTasks based on the leading digits of the log’s IP:

public class MyPartitioner extends Partitioner<Text, DemoEntity> {
@Override
public int getPartition(Text text, DemoEntity demoEntity, int numPartitions) {
    // 111.224.80.24 - - [17/Mar/2021:03:17:49 +0000] "GET /dictionary/gender HTTP/1.1" 200 405 "http://www.renfei.net/index.html" "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.100 Safari/537.36"
    // Take the IP
    String ip = text.toString().split(" ")[0];
    // Say we partition by IP prefix
    if (ip.startsWith("192.")) {
        return 0;
    } else if (ip.startsWith("10.10.")) {
        return 1;
    } else if (ip.startsWith("10.0.")) {
        return 2;
    } else {
        return 3;
    }
}

To use our custom MyPartitioner, set it on the job:

Job job = Job.getInstance(new Configuration());
job.setJarByClass(PartitionerDriver.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(DemoEntity.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(DemoEntity.class);
// Here we only demonstrate using a custom partitioner
// Set ReduceTasks to 4, since our partitions are 0, 1, 2, 3
job.setNumReduceTasks(4);
job.setPartitionerClass(MyPartitioner.class);

Sorting with WritableComparable

In the earlier examples the input we received was already sorted — both MapTask and ReduceTask sort by Key whether you want it or not. Hadoop sorts regardless, lexicographically by default, using quicksort.

We built our own bean back in Hadoop for Beginners (12): Hadoop’s Writable Classes. To make that bean sortable, implement WritableComparable and override compareTo():

public class DemoEntity implements WritableComparable<DemoEntity> {
    private String ip;
    private String path;
    private int port;
    /**
     * Serialization
     *
     * @param dataOutput the data sink the framework gives us
     * @throws IOException
     */
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(ip);
        dataOutput.writeUTF(path);
        dataOutput.writeInt(port);
    }
    /**
     * Deserialization
     *
     * @param dataInput the data source the framework gives us
     * @throws IOException
     */
    @Override
    public void readFields(DataInput dataInput) throws IOException {
        ip = dataInput.readUTF();
        path = dataInput.readUTF();
        port = dataInput.readInt();
    }
    // Getters/setters omitted ....
    /**
     * Sorting support
     *
     * @param o
     * @return
     */
    @Override
    public int compareTo(DemoEntity o) {
        // Say we sort by port
        return Integer.compare(o.getPort(), this.port);
    }
}

The Combiner

Combiner is a third kind of component besides Mapper and Reducer — its parent class is Reducer, but it isn’t quite the same thing. A Combiner runs on the node hosting each MapTask and summarizes that MapTask’s output locally, cutting network I/O much like compression would. Combiners are off by default, because they alter the MapTask’s output: using one is only valid if it doesn’t change the business result. Decide based on your own scenario.

To define and use one, extend Reducer, override the reduce method, then set job.setCombinerClass(MyCombiner.class);. This is beginner-level material and, to be honest, I’m too lazy to write it out — I’ve been working on this tutorial for half a month — so no demo here.

Grouping with GroupingComparator

Grouping is easy to grasp. Say you have sales order data and want the highest-value order for each month: you’d group by month. Write a custom class extending WritableComparator and override compare(). I’m skipping that demo too — look it up when you need it.