How to Write an MR Program

How to Write a MapReduce Program

0. Dependency Import

<dependencies>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-slf4j-impl</artifactId>
        <version>2.12.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.1.3</version>
    </dependency>
</dependencies>

1. Write the Driver Class

public class WordcountDriver {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

		// 1 Get configuration information and obtain the job object
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);

		// 2 Associate the jar of this Driver program
		job.setJarByClass(WordcountDriver.class);

		// 3 Associate the jars of Mapper and Reducer
		job.setMapperClass(WordcountMapper.class);
		job.setReducerClass(WordcountReducer.class);

		// 4 Set the key-value types for Mapper output
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		// 5 Set the final output key-value types
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		// 6 Set the input and output paths
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 7 Submit the job
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

2. Write the Map Class

public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
	Text k = new Text();
	IntWritable v = new IntWritable(1);
	
	@Override
	protected void map(LongWritable key, Text value, Context context)	throws IOException, InterruptedException {
		// 1 Get a line
		String line = value.toString();
		// 2 Split
		String[] words = line.split(" ");
		// 3 Output
		for (String word : words) {		
			k.set(word);
			context.write(k, v);
        //（Ace of Spades, 1）
		}
	}
}

3. Write the Reduce Class

public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

int sum;
IntWritable v = new IntWritable();

	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
		// 1 Accumulate and sum
		sum = 0;
		for (IntWritable count : values) {
			sum += count.get();
		}
		// 2 Output
         v.set(sum);
		context.write(key,v);
	}
}

4. Package and Run

//hadoop jar package name full class name input path output path
[root@hadoop test]# hadoop jar mapreduce.jar WordcountDriver wcinput/ wcoutput

5. View Results

Original file

Ace of Spades 3 of Hearts
Ace of Spades
King of Spades
8 of Hearts
Ace of Spades 3 of Hearts
Ace of Spades
......

Output results

3 of Hearts	114048
4 of Hearts	38016
6 of Hearts	38016
8 of Hearts	76032
Ace of Spades	139392
King of Spades	76032