Hadoop之MapReduce

在Hadoop中，MapReduce起着非常重要的数据处理作用。HDFS作为数据存储的核心部件起着管理文件资源的作用，而MapReduce作为一种编程计算模型，对数据的分析和处理任务进行调度管理。

在MapReduce中，所有的机器分成两个角色，一个叫JobTracker，另外一个叫TaskTracker。同HDFS的分布式概念一样，JobTracker是用于执行总体调度的，一般运行在NameNode机器上，而TaskTracker则负责在节点DataNode上执行具体的分布式任务的计算。海量的数据的计算任务就被许多个TaskTracker将计算量分布在每一个节点上，从而最大程度的利用资源。

更多Hadoop相关信息见Hadoop 专题页面 http://www.linuxidc.com/topicnews.aspx?tid=13

在MapReduce的入门也有一个HelloWord的例子，源代码如下：

1. package com.ant.ren.hadoop.demo;  
2.   
3. import java.io.IOException;  
4. import java.util.StringTokenizer;  
5.   
6. import org.apache.hadoop.conf.Configured;  
7. import org.apache.hadoop.fs.Path;  
8. import org.apache.hadoop.io.IntWritable;  
9. import org.apache.hadoop.io.LongWritable;  
10. import org.apache.hadoop.io.Text;  
11. import org.apache.hadoop.mapreduce.Job;  
12. import org.apache.hadoop.mapreduce.Mapper;  
13. import org.apache.hadoop.mapreduce.Reducer;  
14. import org.apache.hadoop.util.Tool;  
15. import org.apache.hadoop.util.ToolRunner;  
16. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
17. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
18. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
19. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
20.   
21. public class WordCount extends Configured implements Tool{  
22.     /** 
23.     * you can create any class extends Mapper stand for map 
24.     * in map stage, we just split input file content by StringTokenizer default token: whitespace 
25.     * then use word as key, show up count (1) as value 
26.     **/  
27.     public static class Map extends Mapper<LongWritable, Text, Text, IntWritable>{  
28.         private final static IntWritable one = new IntWritable(1);  
29.         private Text word = new Text();  
30.           
31.         public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{  
32.             String line = value.toString();  
33.             StringTokenizer tokenizer = new StringTokenizer(line);  
34.               
35.             while(tokenizer.hasMoreElements()){  
36.                 word.set(tokenizer.nextToken());  
37.                 context.write(word, one);  
38.             }  
39.         }  
40.     }  
41.       
42.     /** 
43.     * you can create any class extends Reducer stand for reduce 
44.     * in reduce stage, we iterate value for the same key (word), calculate the sum of their values 
45.     * then output <word, sum> as reduce result 
46.     **/  
47.     public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable>{  
48.         public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException{  
49.             int sum = 0;  
50.             for(IntWritable val : values)  
51.                 sum += val.get();  
52.             context.write(key, new IntWritable(sum));  
53.         }  
54.     }  
55.       
56.     @Override  
57.     public int run(String[] args) throws Exception {  
58.         Job job = new Job(getConf());  
59.         job.setJarByClass(WordCount.class);  
60.         job.setJobName("Word Count Job");  
61.           
62.         job.setOutputKeyClass(Text.class);  
63.         job.setOutputValueClass(IntWritable.class);  
64.           
65.         job.setMapperClass(Map.class);  
66.         job.setReducerClass(Reduce.class);  
67.           
68.         job.setInputFormatClass(TextInputFormat.class);  
69.         job.setOutputFormatClass(TextOutputFormat.class);  
70.           
71.         FileInputFormat.setInputPaths(job, new Path(args[0]));  
72.         FileOutputFormat.setOutputPath(job, new Path(args[1]));  
73.           
74.         boolean success = job.waitForCompletion(true);  
75.         return success ? 0 : 1;  
76.     }  
77.       
78.     public static void main(String[] args) throws Exception{  
79.         int ret = ToolRunner.run(new WordCount(), args);  
80.         System.exit(ret);  
81.     }     
82. }