MapReduce performs an implicit distributed sort before data reaches each reducer. The following walk-through shows how to exploit this feature to globally order a dataset by an integer column.

Shuffle & Sort Internals

The framework guarantees that every reducer receives its partition already sorted by key. The steps below highlight the critical points:

1. Map phase
   • Each split is processed by one mapper. Output is first buffered in memory (mapreduce.task.io.sort.mb). When the buffer reaches mapreduce.map.sort.spill.percent (default 0.8) the thread spills sorted runs to local disk.
   • During the spill, data is partitioned (hash of key mod R) and optionally combined.
   • After the last record, all spill files are merged into one per partition, still sorted. Compression can be enabled with mapreduce.map.output.compress=true.
2. Reduce phase
   • Each reducer fetches its partitions from every mapper. If the aggregated size fits in memory (mapreduce.reduce.shuffle.input.buffer.percent) it is kept in RAM; otherwise it is merged on disk.
   • The final merge produces a single sorted stream that is fed directly to the reduce function.

Lab Environment

$ start-dfs.sh && start-yarn.sh
$ mkdir -p /data/sortdemo
$ unzip hadoop2lib.zip -d /data/sortdemo
$ hadoop fs -mkdir -p /sortdemo/in
$ hadoop fs -put /data/sortdemo/goods_click.txt /sortdemo/in

Ensure the input file uses comma-separated fields with no trailing spaces.

Driver & Job Configuration

package demo.sort;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class GlobalSortJob extends Configured implements Tool {

  public static class TokenizerMapper
      extends Mapper<LongWritable, Text, IntWritable, Text> {
    private final IntWritable count = new IntWritable();
    private final Text product = new Text();

    @Override
    protected void map(LongWritable offset, Text line, Context ctx)
        throws java.io.IOException, InterruptedException {
      String[] fields = line.toString().split(",");
      count.set(Integer.parseInt(fields[1]));
      product.set(fields[0]);
      ctx.write(count, product);
    }
  }

  public static class IdentityReducer
      extends Reducer<IntWritable, Text, IntWritable, Text> {
    @Override
    protected void reduce(IntWritable key, Iterable<Text> values, Context ctx)
        throws java.io.IOException, InterruptedException {
      for (Text val : values) {
        ctx.write(key, val);
      }
    }
  }

  @Override
  public int run(String[] args) throws Exception {
    if (args.length != 2) {
      System.err.println("Usage: GlobalSortJob <in> <out>");
      return -1;
    }

    Job job = Job.getInstance(getConf(), "global-sort");
    job.setJarByClass(GlobalSortJob.class);

    job.setMapperClass(TokenizerMapper.class);
    job.setReducerClass(IdentityReducer.class);

    job.setMapOutputKeyClass(IntWritable.class);
    job.setMapOutputValueClass(Text.class);

    job.setOutputKeyClass(IntWritable.class);
    job.setOutputValueClass(Text.class);

    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

    return job.waitForCompletion(true) ? 0 : 1;
  }

  public static void main(String[] args) throws Exception {
    System.exit(ToolRunner.run(new GlobalSortJob(), args));
  }
}

Logging Configuration

Create src/main/resources/log4j.properties:

log4j.rootLogger=INFO, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{2}: %m%n

Compile & Run

$ mvn package
$ hadoop jar target/sortdemo.jar demo.sort.GlobalSortJob \
      /sortdemo/in /sortdemo/out
$ hadoop fs -cat /sortdemo/out/part-r-00000

The output will list products in ascending order of their click counts, demonstrating the built-in global sort capability of MapReduce.