Hadoop MapReduce Tutorial

Hadoop MapReduce Tutorial

from 

Hadoop 0.21 Documentation

The Hadoop Documentation provides the information you need to get started using Hadoop, the Hadoop Distributed File System (HDFS), and Hadoop on Demand (HOD).

MapReduce Tutorial

1 Purpose

This document comprehensively describes all user-facing facets of the Hadoop MapReduce framework and serves as a tutorial.

2 Pre-requisites

Ensure that Hadoop is installed, configured and is running. More details:

•  for first-time users.
•  for large, distributed clusters.

3 Overview

2011-03-03 Thu

• Hadoop MapReduce is a software framework for easily writing applications which process vast amounts of data (multi-terabyte data-sets) in-parallel on large clusters (thousands of nodes) of commodity hardware in a reliable, fault-tolerant manner.
• A MapReduce job usually splits the input data-set into independent chunks which are processed by the map tasks in a completely parallel manner. The framework sorts the outputs of the maps, which are then input to the reduce tasks. Typically both the input and the output of the job are stored in a file-system. The framework takes care of scheduling tasks, monitoring them and re-executes the failed tasks.
• Typically the compute nodes and the storage nodes are the same, that is, the MapReduce framework and the Hadoop Distributed File System (see ) are running on the same set of nodes. This configuration allows the framework to effectively schedule tasks on the nodes where data is already present, resulting in very high aggregate bandwidth across the cluster.
• The MapReduce framework consists of a single master JobTracker and one slave TaskTracker per cluster-node. The master is responsible for scheduling the jobs' component tasks on the slaves, monitoring them and re-executing the failed tasks. The salves execute the tasks as directed by the master.
• Minimally, applications specify the input/output locations and supply map and reduce functions via implementations or appropriate interfaces and/or abstract-classes. These, and other job parameters, comprise the job configuration. The Hadoop job client then submits the job (jar/executable etc.) and configuration to the JobTracker which then assumes the responsibility of distributing the software/configuration to the slaves, scheduling tasks and monitoring them, providing status and diagnostic information to the job-client.
• Although the Hadoop framework is implemented in Java^TM, MapReduce application need not be written in Java.

4 Inputs and Outputs

• The MapReduce framework operates exclusively on  pairs, that is, the framework views the input to the job as a set of  pairs and produces a set of  pairs as the output of the job, conceivably of different types.
• The key and value classes have to be serializable by the framework and hence need to implement the Writable interface. Additionally, the key classes have to implement the WritableComparable interface to facilitate sorting by the framework.
• Input and Output types of a MapReduce job:

(input) -> map -> -> combine *-> -> reduce -> (output)

Note that the combine phrase may run zero or more times in this process.

5 Example: WordCount v1.0

• Before we jump into the details, lets walk through an example MapReduce application to get a flavor for how they work.
• WordCount is a simple application that counts the number of occurrences of each word in a given input set.
• This example works with a pseudo-distributed (Single Node Setup) or fully-distributed (Cluster Setup) Hadoop installation.

5.1 Source Code

2011-03-04 Fri

1. package org.myorg;
2.
3. import java.io.IOException;
4. import java.util.*;
5.
6. import org.apache.hadoop.fs.Path;
7. import org.apache.hadoop.conf.*;
8. import org.apache.hadoop.io.*;
9. import org.apache.hadoop.mapreduce.*;
10. import org.apache.hadoop.mapreduce.lib.input.*;
11. import org.apache.hadoop.mapreduce.lib.output.*;
12. import org.apache.hadoop.util.*;
13.
14. public class WordCount extends Configured implements Tool {
15.
16. public static class Map
17. extends Mapper {
18. private final static IntWritable one = new IntWritable(1);
19. private Text word = new Text();
20.
21. public void map(LongWritable key, Text value, Context context)
22. throws IOException, InterruptedException {
23. String line = value.toString();
24. StringTokenizer tokenizer = new StringTokenizer(line);
25. while (tokenizer.hasMoreTokens()) {
26. word.set(tokenizer.nextToken());
27. context.write(word, one);
28. }
29. }
30. }
31.
32. public static class Reduce
33. extends Reducer {
34. public void reduce(Text key, Iterable values,
35. Context context) throws IOException, InterruptedException {
36.
37. int sum = 0;
38. for (IntWritable val : values) {
39. sum += val.get();
40. }
41. context.write(key, new IntWritable(sum));
42. }
43. }
44.
45. public int run(String [] args) throws Exception {
46. Job job = new Job(getConf());
47. job.setJarByClass(WordCount.class);
48. job.setJobName("wordcount");
49.
50. job.setOutputKeyClass(Text.class);
51. job.setOutputValueClass(IntWritable.class);
52.
53. job.setMapperClass(Map.class);
54. job.setCombinerClass(Reduce.class);
55. job.setReducerClass(Reduce.class);
56.
57. job.setInputFormatClass(TextInputFormat.class);
58. job.setOutputFormatClass(TextOutputFormat.class);
59.
60. FileInputFormat.setInputPaths(job, new Path(args[0]));
61. FileOutputFormat.setOutputPath(job, new Path(args[1]));
62.
63. boolean success = job.waitForCompletion(true);
64. return success ? 0 : 1;
65. }
66.
67. public static void main(String[] args) throws Exception {
68. int ret = ToolRunner.run(new WordCount(), args);
69. System.exit(ret);
70. }
71. }
72.

5.2 Usage

(1) Compile WordCount.java and create a jar:

$ cat hadoopjar
# hadoopjar: compile WordCount.java and create a jar
# written by mht on Mar 4, 2011

cd hadoop-0.20.2

mkdir wordcount_classes

export HADOOP_HOME=~/hadoop-0.20.2
export HADOOP_VERSION="0.20.2"

#echo $HADOOP_VERSION

javac -classpath ${HADOOP_HOME}/hadoop-${HADOOP_VERSION}-core.jar:
${HADOOP_HOME}/hadoop-${HADOOP_VERSION}-mapred.jar:
${HADOOP_HOME}/hadoop-${HADOOP_VERSION}-hdfs.jar
-d wordcount_classes wordcount_src/WordCount.java

export JAVA_HOME=/usr/local/jdk1.6.0_17
export CLASSPATH=${JAVA_HOME}/lib/tools.jar

jar -cvf wordcount_jar/wordcount.jar -C wordcount_classes/ .
$

(2) Sample text-files as input:

$ ls input
file01 file02

$ cat input/file01
Hello World Bye World

$ cat input/file02
Hello Hadoop Goodbye Hadoop

(3) Run the application:

$ bin/hadoop jar wordcount_jar/wordcount.jar org.myorg.WordCount input output

(4) Output:

$ cat output/part-r-00000
Bye 1
Goodbye 1
Hadoop 2
Hello 2
World 2

5.3 Bundling a data payload with your application

• Applications can specify a comma-separated list of paths which would be present in the current working directory of the task using the option -files. The -libjars option allows applications to add jars to the classpaths of the maps and reduces. The option -archives allows them to pass comma separated list of archives as arguments. These archives are unarchived and a link with name of the archive is created in the current working idrectory of tasks.
• The mechanism that provides this functionality is called the distributed cache. More detains about the command line options surrounding job launching and control of the distributed cache are avaiable at [[][Hadoop Commands Guide]].
• Hadoop ships with some example code in a jar precompiled for you; one of these is (another) wordcount program. Here's an example invocation of the wordcount example with -libjars, -files and -archives;

hadoop jar hadoop-examples.jar wordcount -files cachefile.txt
-libjars mylib.jar -archives myarchive.zip input output

Here, myarchive.zip will be placed and unzipped into a directory by the name "myarchive.zip".

• Users can specify a different symbolic name for files and archives passed through -files and -archives option, using #.

For example,

hadoop jar hadoop-examples.jar wordcount
-files dir1/dict.txt#dict1, dir2/dict.txt#dict2
-archives mytar.tgz#tgzdir input output

Here, the files dir1/dict.txt and dir2/dict.txt can be accessed by tasks using the symbolic names dict1 and dict2 respectively. And the archive mytar.tgz will be placed and unarchived into a directory by the name tgzdir.

5.4 Walk-through

2011-03-06 Sun
This section describes the operation of the WordCount application shown earlier in this tutorial.

The Mapper implementation (lines 16-30), via the map method (lines 21-29), processes one line at a time, as provided by the specified TextInputFormat. It then splits the line into tokens separated by whitespaces, via theStringTokenizer, and emits a key-value pair of <, 1>.

For the given sample input the first map emits:

The second map emits:

WordCount also specifies a combiner (line 54). Hence, the output of each map is passed through the local combiner (which is same as the Reducer as per the job configuration) for local aggregation, after being sorted on the keyss.

The output of the first map:

The output of the second map:

The Reducer implementation (lines 32-43), via the reduce method (lines 34-42) just sums up the values, which are the occurence counts for each key (i.e. words in this example).

Thus the output of the job is:

The run method specifies various facets of the job, such as the input/output paths (passed via the command line), key/value types, input/output formats etc., in the Job. It then calls the Job.waitForCompletion() (line 63) to submit the job to Hadoop and monitor its progress.

6 MapReduce - User Interfaces

This section provides a reasonalbe amout of detail on every user-facing aspect of the MapReduce framework. This should help users implement, configure and tune their jobs in a fine-grained manner.

2011-03-06 Sun

6.1 Payload

Applications typically extend the Mapper and Reducer classes to provide the map and reduce methods. Thsese form the core of the job.

6.1.1 Mapper

Mapper maps input key/value pairs to a set of intermediate key/value pairs. Maps are the individual tasks that transform input records into intermediate records. The Hadoop MapReduce framewrok spawns one map task for eachInputSplit generated by the InputFormat for the job.

An InputSplit is a logical representation of a unit of input work for a map task; e.g., a filename and a byte range within that file to process. The InputFormat is responsible for enumerating the InputSplits, and producting aRecordReader which will turn those logical work units into actual physical input records.

• Overall, Mapper implementations are specified in the Job, a client-side class that describes the job's configuration and interfaces with the cluster on behalf of the client program. The Mapper itself then is instantiated in the running job, and is passed a MapContext object which it can use to configure itself. The Mapper contains a run() method which calls its setup() method once, its map() method for each input record, and finally its cleanup()method. All of these methods (including run() itself) can be overridden with your own code. If you do not override any methds (leaving even map as-is), it will act as the identity function, emmitting each input record as a separate output.

(1) The Context object allows the mapper to interact with the rest of the Hadoop system. It includes configuration data for the job, as well as interfaces which allow it to emit output. The getConfiguration() method returns aConfiguration which contains configuration data for your program. You can set arbitrary (key, value) pairs of configuration data in your Job, e.g. with Job.getConfiguration().set("myKey", "myval"), and then retrieve this data in your mapper with Context.getConfiguration().get("myKey"). This sort of functionality is typically done in the Mapper's setup() method.

(2) The Mapper.run() method then calls map(KeyInType, ValInType, Context) for each key/value pair in the InputSplit for that task. Note that in the WordCount program's map() method, we then emit our output data via theContext argument, using its write() method.

(3) Applications can then override the Mapper's Cleanup() method to perform any required teardown operations.

(4) Output pairs are collected with calls to Context.write(KeyOutType, ValOutType).

(5) Applications can also use the Context to report progress, set application-level status messages and update Counters, or just indicate that they are alive.

(6) All intermediate values associated with a given output key are subsequently grouped by the framework, and passed to the Reducer(s) to determine the final output. Users can control the grouping by specifying a Comparator viaJob.setGroupingComparatorClass(class).

(7) The Mapper outputs are sorted and partitioned per Reducer. The total number of partitions is the same as the number of reduce tasks for the job. Users can control which keys (and hence records) go to which Reducer by implementing a custom Partitioner.

(8) Users can optionally specify a combiner, via Job.setCombinerClass(Class), to perform local aggregation of the intermediate outputs, which helps to cut down the amount of data transferred from the Mapper to the Reducer.

(9) The intermediate, sorted outputs are always stored in a simle (key-len, key, value-len, value) format. Applications can control if, and how, the intermediate outputs are to be compressed and the CompressionCodec to be used via the Job.

How Many Maps?

• The number of maps is usually driven by the total size of the inputs, that is, the total number of blocks of the input files.
• The right level of parallelism for maps seems to be around 10-100 maps per-node, although it has been set up to 300 maps for very cpu-light map tasks. Task setup takes awhile, so it is best if the maps take at least a minute to execute.
• Thus, if you expect 10TB of input data and have a blocksize of 128MB, you'll end up with 82,000 maps, unless the mapreduce.job.maps parameter (which only provides a hint to the framework) is used to set it even higher. Ultimately, the number of tasks is controlled by the number of splits returned by the InputFormat.getSplits() method.

6.1.2 Reducer

2011-03-07 Mon

Reducer reduces a set of intermediate values which share a key to a (usually smaller) set of values. The number of reduces for the job is set by the user via Job.setNumReduceTasks(int).

The API of Reducer is very similar to that of Mapper; there's a run() method that receives a Context containing the job's configuration as well as interfacing methods that return data from the reducer itself back to the framework. Therun() method calls setup() once, reduce() once for each key associated with the reduce task, and cleanup() once at the end. Each of these methods can access the job's configuration data by using Context.getconfiguration().

The heart of Reducer is its reduce() method. This is called once per key; they second argument is an Iterable which returns all the values associated with that key. The Reducer should emit its final output (key, value) pairs with theContext.write() method. It may emit 0, 1, or more (key, value) pairs for each input.

Reducer has 3 primary phases: shuffle, sort and reduce.

• Shuffle 
  Input to the Reducer is the sorted output of the mappers. In this phase the framework fetches the relevant partition of the output of all the mappers, via HTTP.

• Sort 
  The framework groups Reducer inputs by keys (since different mappers may have output the same key) in this stage.

  The shuffle and sort phase occur simultaneously; while map-outputs are being fetched they are merged.

• Secondary Sort 
  

  If equivalence rules for grouping the intermediate keys are required to be different from those for grouping keys before reduction, then one may specify a Comarator via Job.setGroupingComparatorClass(Class). Since this can be used to control how intermediate keys are grouped, these can be used in conjunction to simulate secondary sort on values.

• Reduce 
  In this phase the reduce(MapOutKeyType, Iterable , Context) method is called for each  pair in the grouped inputs.

  The output of the reduce task is typically written to the FileSystem via Context.write(ReduceKeyType, ReduceOutValType).

  Applications can use the Context to report progress, set application-level status messages and update counters, or just indicate that they are alive.

  The output of the Reducer is not sorted.

6.1.3 How Many Reduces?

The right number of reduces seems to be 0.95 or 1.75 multiplied by ( * mapreduce.tasktracker.reduce.tasks.maximum*).

With 0.95 all of the reduces can launch immediately and start transfering map outputs as the maps finish. With 1.75 the faster nodes will finish their first round of reduces and launch a second wave of reduces doing a much better job of load balancing.

Increasing the number of reduces increases the framework overhead, but increases load balancing and lowers the cost of failures.

The scaling factors above are slightly less than whole numbers to reserve a few reduce slots in the framework for speculative-tasks and failed tasks.

6.1.4 Reducer NONE

It is legal to set the number of reduce-tasks to zero if no reduction is desired.

In this case the outputs of the map-tasks go directly to the FileSystem, into the output path set by setOutputPath(Path). The framework does not sort the map-outputs before writing them out to the FileSystem..

6.1.5 Mark-Reset

While applications iterate through the values for a given key, it is possible to mark the current position and later reset the iterator to this position and continue the iteration process. The corresponding methods are mark() and reset().

mark() and reset() can be called any number of times during the iteration cycle. The reset() method will reset the iterator to the last record before a call to the previous mark().

This functionality is avaiable only with the new context based reduce iterator.

6.1.6 Partitioner

Partitioner partitions the key space.

Partitioner controls the partitioning of the keys of the intermediate map-outputs. The key (or a subset of the key) is used to derive the partition, typically by a hash function. The total number of partitions is the same as the number of reduce tasks for the job. Hence this controls which of the m reduce tasks the intermediate key (and hence the record) is sent to for the reduction.

HashPatitioner is the default Partitioner.

6.2 Reporting Progress

Via the mapper or reducer's Context, MapReduce applications can report progress, set application-level status messages and update Counters.

Mapper and Reducer implementations can use the Context to report progress or just indicate that they are alive.

In scenarios where the application takes a significant amount of time to process individual key/value pairs, this is crucial since the framework might assume that the task has time-out and kill that task. Another way to avoid this is to set the configuration parameter mapreduce.task.timeout to a high-enough value (or even set it to zero for no time-outs).

Applications can also update Counters using the Context.

Hadoop MapReduce comes bundled with a library of generally useful mappers, reducers, and partitioners in the org.apache.hadoop.mapreduce.lib package.

6.3 Job configuration

2011-03-07 Mon
The Job represents a MapReduce job configuration. The actual state for this object is written to an underlying instance of Configuration.

Job is the primary interface for a user to describe a MapReduce job to the Hadoop framework for execution.

The Job is typically used to specify the Mapper, combiner (if any), Partitioner, Reducer, InputFormat, OutputFormat and OutputCommitter implementations. Job also indicates the set of input files (setInputPaths(Job, Path…))/*addInputPath(Job,Path))* and (setInputPaths(Job, String)*/*(addInputPaths(Job,String)) and where the output files should be written (setOutputPath(Path)).

Of course, users can use Job.getConfiguration() to get access to the underlying configuration state, and can then use set(String, String)/get(String,String) to set/get arbitrary parameters needed by applications. However, use theDistributedCache for large amounts of (read-only) data.

6.4 Task Execution & Environment

2011-03-07 Mon

The TaskTracker executes the Mapper/Reducer task as a child process in a separate jvm.

The child-task inherits the environment of the parent TaskTracker. The user can specify additional

6.4.1 Configuring Memory Requirements For A Job

6.4.2 Map Parameters

6.4.3 Shuffle/Reduce Parameters

6.4.4 Directory Structure

6.4.5 Task JVM Reuse

6.4.6 Task Logs

6.4.7 Distributing Libraries

6.5 Job Submission and Monitoring

2011-03-07 Mon

The Job is the primary interface by which user-job interacts with the JobTracker.

Job provides facilities to submit jobs, track their progress, access component-tasks' reports and logs, get the MapReduce cluster's status information and so on.

The job submission process involves:

1. Checking the input and output specifications of the job.
2. Computing the InputSplit values for the job.
3. Setting up the requisite accounting information for the DistributedCache of the job, if necessary.
4. Copying the job's jar and configuration to the MapReduce sstem directory on the FileSystem.
5. Submitting the job to the JobTracker and optionally monitoring it's status.

Normally the user creates the application, describes various facets of the job via Job, and then uses the waitForCompletion() method to submit the job and monitor its progress.

6.5.1 Job Control

6.5.2 Job Authorization

6.6 Job Input

2011-03-07 Mon

InputFormat describes the input-specification for a MapReduce job.

The MapReduce framework relies on the InputFormat of the job to:

1. Validate the input-specification of the job.
2. Split-up the input file(s) into logical InputSplit instances, each of which is then assigned to an individual Mapper.
3. Provide the RecordReader implementation used to glean input records from the logical InputSplit for processing by the Mapper.

The default behavior of file-based InputFormat implementations, typically sub-classes of FileInputFormat, is to split the input into logical InputSplit instances based on the total size, in bytes, of the input files. However, theFileSystem blocksize of the input files is treated as an upper bound for input splits. A lower bound on the split size can be set via mapreduce.input.fileinputformat.split.minsize.

TextInputFormat is the default InputFormat.

If TextInputFormat is the InputFormat for a given job, the framework detects input-files with the .gz extensions and automatically decompresses them using the appropriate CompressionCodec. However, it must be noted that compressed files with the above extensions cannot be split and each compressed file is processed in its entirety by a single mapper.

6.6.1 InputSplit

InputSplit represents the data to be processed by an individual Mapper.

Typically InputSplit presents a byte-oriented view of the input, and it is the responsibility of RecordReader to process and present a record-oriented view.

FileSplit is the default InputSplit. It sets mapreduce.map.input.file to the path of the input file for the logical split.

6.6.2 RecordReader

RecordReader reads  pairs from an InputSplit.

Typically the RecordReader converts the byte-oriented view of the input, provided by the InputSplit, and presents a record-oriented to the Mapper implementations for processing. RecordReader thus assumes the responsibility of processing record boundaries and presents the tasks with keys and values.

6.7 Job Output

2011-03-09 Wed

OutputFormat describes the output-specification for a MapReduce job.

The MapReduce framework relies on the OutputFormat of the job to:

1. Validate the output-specification of the job; for example, check that the output directory doesn't already exist.
2. Provide the RecordWriter implementation used to write the output files of the job. Output files are stored in a FileSystem.

TextOutputFormat is the default OutputFormat.

RecodWriter writes the output  pairs to an output file. RecordWriter implementations write the job outputs to the FileSystem.

6.8 Other Useful Features

2011-03-09 Wed

6.8.1 Submitting Jobs to Queues

6.8.2 Counters

6.8.3 DistributedCache

6.8.4 Tool

The Tool interface supports the handling of generaic Hadoop command-line options.

6.8.5 IsolationRunner

IsolationRunner is a utility to help debug MapReduce programs.

6.8.6 Profiling

Profiling is a utility to get a representative (2 or 3) sample of build-in java profiler for a sample of maps and reduces.

6.8.7 Debugging

6.8.8 JobControl

6.8.9 Data Compression

6.8.10 Skipping Bad Records

7 Example: WordCount v2.0

2011-03-09 Wed
Here is a more complete WordCount which uses many of the features provided by the MapReduce framework we dicussed so far.

This example needs the HDFS to be up and running, especially for the DistributedCache-related features. Hence it only works with a pseudo-distributed (Single Node Setup) or fully-distributed (Cluster Setup) Hadoop installtation.

7.1 Source Code

2011-03-09 Wed

1. package org.myorg;
2.
3. import java.io.*;
4. import java.util.*;
5.
6. import org.apache.hadoop.fs.Path;
7. import org.apache.hadoop.filecache.DistributedCache;
8. import org.apache.hadoop.conf.*;
9. import org.apache.hadoop.io.*;
10. import org.apache.hadoop.mapreduce.*;
11. import org.apache.hadoop.mapreduce.lib.input.*;
12. import org.apache.hadoop.mapreduce.lib.output.*;
13. import org.apache.hadoop.util.*;
14.
15. public class WordCount2 extends Configured implements Tool {
16.
17. public static class Map
18. extends Mapper {
19.
20. static enum Counters { INPUT_WORDS }
21.
22. private final static IntWritable one = new IntWritable(1);
23. private Text word = new Text();
24.
25. private boolean caseSensitive = true;
26. private Set patternsToSkip = new HashSet();
27.
28. private long numRecords = 0;
29. private String inputFile;
30.
31. public void setup(Context context) {
32. Configuration conf = context.getConfiguration();
33. caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
34. inputFile = conf.get("mapreduce.map.input.file");
35.
36. if (conf.getBoolean("wordcount.skip.patterns", false)) {
37. Path[] patternsFiles = new Path[0];
38. try {
39. patternsFiles = DistributedCache.getLocalCacheFiles(conf);
40. } catch (IOException ioe) {
41. System.err.println("Caught exception while getting cached files: "
42. + StringUtils.stringifyException(ioe));
43. }
44. for (Path patternsFile : patternsFiles) {
45. parseSkipFile(patternsFile);
46. }
47. }
48. }
49.
50. private void parseSkipFile(Path patternsFile) {
51. try {
52. BufferedReader fis = new BufferedReader(new FileReader(
53. patternsFile.toString()));
54. String pattern = null;
55. while ((pattern = fis.readLine()) != null) {
56. patternsToSkip.add(pattern);
57. }
58. } catch (IOException ioe) {
59. System.err.println("Caught exception while parsing the cached file '"
60. + patternsFile + "' : " + StringUtils.stringifyException(ioe));
61. }
62. }
63.
64. public void map(LongWritable key, Text value, Context context)
65. throws IOException, InterruptedException {
66. String line = (caseSensitive) ?
67. value.toString() : value.toString().toLowerCase();
68.
69. for (String pattern : patternsToSkip) {
70. line = line.replaceAll(pattern, "");
71. }
72.
73. StringTokenizer tokenizer = new StringTokenizer(line);
74. while (tokenizer.hasMoreTokens()) {
75. word.set(tokenizer.nextToken());
76. context.write(word, one);
77. context.getCounter(Counters.INPUT_WORDS).increment(1);
78. }
79.
80. if ((++numRecords % 100) == 0) {
81. context.setStatus("Finished processing " + numRecords
82. + " records " + "from the input file: " + inputFile);
83. }
84. }
85. }
86.
87. public static class Reduce
88. extends Reducer {
89. public void reduce(Text key, Iterable values,
90. Context context) throws IOException, InterruptedException {
91.
92. int sum = 0;
93. for (IntWritable val : values) {
94. sum += val.get();
95. }
96. context.write(key, new IntWritable(sum));
97. }
98. }
99.
100. public int run(String[] args) throws Exception {
101. Job job = new Job(getConf());
102. job.setJarByClass(WordCount2.class);
103. job.setJobName("wordcount2.0");
104.
105. job.setOutputKeyClass(Text.class);
106. job.setOutputValueClass(IntWritable.class);
107.
108. job.setMapperClass(Map.class);
109. job.setCombinerClass(Reduce.class);
110. job.setReducerClass(Reduce.class);
111.
112. // Note that these are the default.
113. job.setInputFormatClass(TextInputFormat.class);
114. job.setOutputFormatClass(TextOutputFormat.class);
115.
116. List other_args = new ArrayList();
117. for (int i=0; i < args.length; ++i) {
118. if ("-skip".equals(args[i])) {
119. DistributedCache.addCacheFile(new Path(args[++i]).toUri(),
120. job.getConfiguration());
121. job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
122. } else {
123. other_args.add(args[i]);
124. }
125. }
126.
127. FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
128. FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));
129.
130. boolean success = job.waitForCompletion(true);
131. return success ? 0 : 1;
132. }
133.
134. public static void main(String[] args) throws Exception {
135. int res = ToolRunner.run(new Configuration(), new WordCount2(), args);
136. System.exit(res);
137. }
138. }

7.2 Sample Runs

2011-03-10 Thu

Sample text-files as input:

$ bin/hadoop fs -ls /user/joe/wordcount/input/
/user/joe/wordcount/input/file01
/user/joe/wordcount/input/file02

$ bin/hadoop fs -cat /user/joe/wordcount/input/file01
Hello World, Bye World!

$ bin/hadoop fs -cat /user/joe/wordcount/input/file02
Hello Hadoop, Goodbye to hadoop.

Run the application:

$ bin/hadoop jar /user/joe/wordcount.jar org.myorg.WordCount2 /user/joe/wordcount/input /user/joe/wordcount/output

Output:

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
Bye 1
Goodbye 1
Hadoop, 1
Hello 2
World! 1
World, 1
hadoop. 1
to 1

Notice that the inputs differ from the first version we looked at, and how they affect the outputs.

Now, lets plug-in a pattern-file which lists the word-patterns to be ignored, via the DistributedCache.

$ hadoop fs -cat /user/joe/wordcount/patterns.txt
\.
\,
\!
to

Run it again, this time with more options:

$ bin/hadoop jar /user/joe/wordcount.jar org.myorg.WordCount2 -Dwordcount.case.sensitive=true /user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

As expected, the output:

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
Bye 1
Goodbye 1
Hadoop 1
Hello 2
World 2
hadoop 1

Run it once more, this time switch-off case-sensitivity:

$ bin/hadoop jar /user/joe/wordcount.jar org.myorg.WordCount2 -Dwordcount.case.sensitive=false /user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

Sure enough, the output:

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
bye 1
goodbye 1
hadoop 2
hello 2
world 2

7.3 Highlights

2011-03-10 Thu
The second version of WordCount improves upon the previous one by using some features offered by the MapReduce framework:

• Demonstrates how applications can access configuration parameters in the setup method of the Mapper (and Reducer) implementations (lines 31-48).

-Demonstrates how the DistributedCache can be used to distribute read-only data needed by the jobs. Here it allows the user to specify word-patterns to skip while counting (line 119).

• Demonstrates the utility of the Tool interface and the GenericOptionsParser to handle generic Hadoop command-line options (line 135).
• Demonstrates how applications can use Counters (line 77) and how they can set application-specific status information via the Context instance passed to the map (and reduce) method (line 81).

Author: mahaitao

Date: 2011-03-10 14:55:39 EST

HTML generated by org-mode 7.4 in emacs 23