Programming with Hadoop

1. Two Core Operations
- 1.1. Manipulating HDFS Files
- 1.2. TODO MapReduce Paradigm to Program Computation Tasks

Programming Hadoop with Java is indeed writing Java code to call Hadoop to perform operations.

1. Two Core Operations

1.1. Manipulating HDFS Files

HDFS is Hadoop’s storage system. Through Java APIs, we can perform file IOs as local files, including creation, read, write, deletion.

The core of manipulation is Configuration class and FileSystem class, the former loads Hadoop configurations and the latter provides an entry to operating HDFS file system, acting like a handle.

1.2. TODO `MapReduce` Paradigm to Program Computation Tasks

Map-reduce is Hadoop’s computation framework, which splits each task into map and reduce.

1.2.1. Mapping Stage

We should override the map() method of Mapper<KeyIn, ValueIn, IntermediateKeyOut, IntermediateValueOut> class to define how to produce intermediate key-value pairs.

The core class is org.apache.hadoop.mapreduce.Mapper<KeyIn, ValueIn, KeyOut, ValueOut> which is an abstract class that users should inherit and override the map() method.

Its lifecycle is:

setup(Context context) which is called only once for preprocessing
map(KeyIn key, ValueIn value, Context context) which is the core processing logic
cleanup(Context context) pst-processing
run(Context context) which control the overall flow, and is often ignored.

And finally, the intermediate output is written through context.write(KeyOut key, ValueOut value).

1.2.2. Reducing Stage

We should override the reduce() method of Reducer<IntermediateKeyIn, IntermediateValueOut, FinalKeyOut, FinalValueOut> class to define how to reduce intermediate results.

The core class is org.apache.hadoop.mapreduce.Reducer<IntermediateKeyIn, IntermediateValueOut, KeyOut, ValueOut>. Similar to Mapper<>, the core method is reduce(IntermediateKeyIn key, Iterable<IntermediateValueIn> values, Context context).

1.2.3. Driver Stage

This is the entrance of the whole program and is responsible for linking all procedure, including

setting up input path
setting up output path
setting which mapper and which reducer.

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;

public class WordCountDriver {
    public static void main(String[] args) throws Exception {
        // 1. Load config
        Configuration conf = new Configuration();
        // 2. Create and name a Job work
        Job job = Job.getInstance(conf, "Word Count");
        // 3. set main entrance
        job.setJarByClass(WordCountDriver.class);
        // 4. set the mapper and reducer that will be used
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        // 5. set the K-V pair type of intermediate output and final output
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 6. set input and output path (from argument)
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 7. submit jobs and wait for completion
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

1.2.4. Framework Support

Writable class and serialization. org.apache.hadoop.io.Writable is responsible for serialization that all key, value types must extend this class.

This class has 2 methods:

void write(DataOutput out) that writes self into binary stream.
void readFields(DataInput in) that fills the field from a binary stream.

Hadoop is shipped with Writable implementation for built-in types, e.g., IntWritable, LongWritable, FloatWritable, DoubleWritable, Text, NullWritable, ArrayWritable, etc.