Writing a MapReduce Job in Java

Analogy categoryAnalogies

Anatomy of a Java MapReduce Job

A complete Hadoop MapReduce job in Java is made of three cooperating pieces: a Driver class containing the main() method that configures and submits the job, a Mapper subclass implementing map(), and a Reducer subclass implementing reduce() (with an optional Combiner, often the same class as the Reducer for associative operations). The Driver constructs a Configuration and a Job object bound to it, wires in the mapper, reducer, combiner, and input/output paths, then calls job.waitForCompletion(true) to submit the job to the cluster (or LocalJobRunner in local mode) and block until it finishes, returning a boolean success flag the driver typically uses to set the process exit code.

🏏

Cricket analogy: Running a franchise involves three cooperating roles: the team owner who sets budget and strategy (Driver), the head coach who trains individual batting technique (Mapper), and the strategist who reviews match footage to refine overall game plans (Reducer), each with a clear separation of duties.

Implementing Mapper and Reducer Classes

Both Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> and Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> are generic classes parameterized by their input and output key/value types, and every type used must implement Hadoop's Writable interface (for values) or WritableComparable (for keys, since keys must be sortable) — common built-ins include Text, IntWritable, LongWritable, DoubleWritable, and NullWritable. Critically, the reducer's KEYIN/VALUEIN types must exactly match the mapper's KEYOUT/VALUEOUT types, since that's the intermediate contract the framework enforces at job submission time; a mismatch, like a mapper emitting IntWritable values while the reducer expects Text, causes a runtime ClassCastException rather than a compile-time error, because the framework can't fully verify generic type parameters across the shuffle boundary.

🏏

Cricket analogy: A bowling coach and a batting coach must agree on exactly which drill format the fielding notes use, since a mismatch, like the bowling coach logging speeds in km/h while the batting coach's system expects mph, only surfaces as a broken report at review time, not before training starts, mirroring the mapper/reducer type contract failing at runtime.

Configuring and Submitting the Job

A typical driver builds the job with Job.getInstance(conf, "job name"), calls setJarByClass() so Hadoop can locate the job's JAR on the classpath, sets setMapperClass(), setCombinerClass() (optional), and setReducerClass(), declares setOutputKeyClass() and setOutputValueClass() for the final output types, optionally overrides setMapOutputKeyClass()/setMapOutputValueClass() when intermediate types differ from final output types, and finally wires FileInputFormat.addInputPath() and FileOutputFormat.setOutputPath() before calling waitForCompletion(). Hadoop refuses to run if the output path already exists, as a safety measure against accidentally overwriting a previous job's results, so driver code (or the shell script invoking it) commonly deletes or timestamps the output directory before resubmitting during iterative development.

🏏

Cricket analogy: Before a match, the umpires check the pitch report, confirm both playing XIs, and verify the toss result before play can begin — skipping any of these checks means the match can't officially start, similar to how a Job object needs its mapper, reducer, and I/O paths all configured before waitForCompletion() can run.

java

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class WordCountDriver extends Configured implements Tool {

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

        Configuration conf = getConf();
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCountDriver.class);

        job.setMapperClass(WordCountMapper.class);
        job.setCombinerClass(WordCountReducer.class);
        job.setReducerClass(WordCountReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.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 {
        int exitCode = ToolRunner.run(new Configuration(), new WordCountDriver(), args);
        System.exit(exitCode);
    }
}

Extending Configured and implementing Tool, then launching via ToolRunner.run(), automatically enables Hadoop's GenericOptionsParser, which parses standard cluster options like -D mapreduce.job.reduces=10, -files, -libjars, and -archives from the command line before your own args[] array is populated — writing a plain main() without ToolRunner silently loses this capability.

Testing and Running on a Cluster

Before submitting to a real cluster, it's standard practice to run the job in local mode (mapreduce.framework.name=local, the default when no cluster config is present) against a small sample dataset, or to use a unit testing library like MRUnit or plain JUnit tests against the Mapper/Reducer classes directly by calling map()/reduce() with mocked Context objects, catching logic bugs cheaply before burning cluster time. Once packaged into a JAR (typically with dependencies shaded in via Maven's shade plugin or a similar mechanism), the job is submitted with hadoop jar mycode.jar com.example.WordCountDriver /input/path /output/path, and progress and errors can be inspected afterward with yarn logs -applicationId <id> or through the YARN ResourceManager web UI.

🏏

Cricket analogy: A batsman practices against a bowling machine in the nets on a small set of deliveries before facing live bowling in a real match, catching technique flaws cheaply before it matters, mirroring local-mode testing before a full cluster run.

A very common runtime error is forgetting to call setMapOutputKeyClass()/setMapOutputValueClass() when the mapper's intermediate output types differ from the final job output types set via setOutputKeyClass()/setOutputValueClass() — Hadoop assumes intermediate and final types match unless told otherwise, and the mismatch surfaces as a ClassCastException deep in the shuffle, often with a confusing stack trace far from the actual misconfiguration.

A Java MapReduce job is composed of a Driver, a Mapper subclass, and a Reducer subclass, with an optional Combiner.
All key and value types must implement Writable (values) or WritableComparable (keys) such as Text or IntWritable.
The reducer's input types must exactly match the mapper's output types, or a runtime ClassCastException occurs.
Job.getInstance, setJarByClass, setMapperClass, setReducerClass, and I/O paths must all be configured before submission.
Extending Tool and using ToolRunner enables standard command-line cluster options via GenericOptionsParser.
Local-mode runs and MRUnit/JUnit tests catch logic bugs cheaply before submitting to a real cluster with hadoop jar.
Forgetting setMapOutputKeyClass/setMapOutputValueClass when intermediate types differ from final types is a common bug.