1 package com.mapreduce;
 2 
 3 
 4 import java.io.IOException;
 5 
 6 import org.apache.commons.lang.StringUtils;
 7 import org.apache.hadoop.io.LongWritable;
 8 import org.apache.hadoop.io.Text;
 9 import org.apache.hadoop.mapreduce.Mapper;
10 
11 /***
12  * 
13  * @author Administrator
14  * 1：4个泛型中，前两个是指定mapper输入数据的类型，KEYIN是输入的key的类型，VALUEIN是输入的value的值
15  *       KEYOUT是输入的key的类型，VALUEOUT是输入的value的值 
16  * 2：map和reduce的数据输入和输出都是以key-value的形式封装的。
17  * 3：默认情况下，框架传递给我们的mapper的输入数据中，key是要处理的文本中一行的起始偏移量，这一行的内容作为value
18  * 4：key-value数据是在网络中进行传递，节点和节点之间互相传递，在网络之间传输就需要序列化，但是jdk自己的序列化很冗余
19  *    所以使用hadoop自己封装的数据类型，而不要使用jdk自己封装的数据类型；
20  *    Long--->LongWritable
21  *    String--->Text    
22  */
23 public class WcMap extends Mapper<LongWritable, Text, Text, LongWritable>{
24 
25     //重写map这个方法
26     //mapreduce框架每读一行数据就调用一次该方法
27     @Override
28     protected void map(LongWritable key, Text value, Context context)
29             throws IOException, InterruptedException {
30         //具体业务逻辑就写在这个方法体中，而且我们业务要处理的数据已经被框架传递进来，在方法的参数中key-value
31         //key是这一行数据的起始偏移量，value是这一行的文本内容
32         
33         //1:切分单词,首先拿到单词value的值,转化为String类型的
34         String str = value.toString();
35         //2:切分单词,空格隔开,返回切分开的单词
36         String[] words = StringUtils.split(str," ");
37         //3:遍历这个单词数组，输出为key-value的格式，将单词发送给reduce
38         for(String word : words){
39             //输出的key是Text类型的，value是LongWritable类型的
40             context.write(new Text(word), new LongWritable(1));
41         }
42         
43         
44     }
45 }

 1 package com.mapreduce;
 2 
 3 import java.io.IOException;
 4 
 5 import org.apache.hadoop.io.LongWritable;
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.mapreduce.Reducer;
 8 
 9 /***
10  * 
11  * @author Administrator
12  * 1:reduce的四个参数，第一个key-value是map的输出作为reduce的输入,第二个key-value是输出单词和次数，所以
13  *      是Text,LongWritable的格式；
14  */
15 public class WcReduce extends Reducer<Text, LongWritable, Text, LongWritable>{
16 
17     //继承Reducer之后重写reduce方法
18     //第一个参数是key，第二个参数是集合。
19     //框架在map处理完成之后，将所有key-value对缓存起来，进行分组，然后传递一个组<key,valus{}>，调用一次reduce方法
20     //<hello,{1,1,1,1,1,1.....}>
21     @Override
22     protected void reduce(Text key, Iterable<LongWritable> values,Context context) 
23             throws IOException, InterruptedException {
24         //将values进行累加操作，进行计数
25         long count = 0;
26         //遍历value的list，进行累加求和
27         for(LongWritable value : values){
28             
29             count += value.get();
30         }
31         
32         //输出这一个单词的统计结果
33         //输出放到hdfs的某一个目录上面，输入也是在hdfs的某一个目录
34         context.write(key, new LongWritable(count));
35     }
36     
37     
38 }

 1 package com.mapreduce;
 2 
 3 import java.io.IOException;
 4 
 5 import org.apache.hadoop.conf.Configuration;
 6 import org.apache.hadoop.fs.Path;
 7 import org.apache.hadoop.io.LongWritable;
 8 import org.apache.hadoop.io.Text;
 9 import org.apache.hadoop.mapreduce.Job;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12 
13 
14 /***
15  * 1:用来描述一个特定的作业
16  *       比如,该作业使用哪个类作为逻辑处理中的map,那个作为reduce
17  * 2:还可以指定该作业要处理的数据所在的路径
18  *        还可以指定改作业输出的结果放到哪个路径
19  * @author Administrator
20  *
21  */
22 public class WcRunner {
23 
24     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
25         //创建配置文件
26         Configuration conf = new Configuration();
27         //获取一个作业
28         Job job = Job.getInstance(conf);
29         
30         //设置整个job所用的那些类在哪个jar包
31         job.setJarByClass(WcRunner.class);
32         
33         //本job使用的mapper和reducer的类
34         job.setMapperClass(WcMap.class);
35         job.setReducerClass(WcReduce.class);
36         
37         //指定reduce的输出数据key-value类型
38         job.setOutputKeyClass(Text.class);
39         job.setOutputValueClass(LongWritable.class);
40         
41         
42         //指定mapper的输出数据key-value类型
43         job.setMapOutputKeyClass(Text.class);
44         job.setMapOutputValueClass(LongWritable.class);
45         
46         //指定要处理的输入数据存放路径
47         FileInputFormat.setInputPaths(job, new Path("hdfs://master:9000/wc/srcdata"));
48         
49         //指定处理结果的输出数据存放路径
50         FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9000/wc/output"));
51         
52         //将job提交给集群运行 
53         job.waitForCompletion(true);
54     }
55     
56 }

 1 package com.mapreduce;
 2 
 3 import java.io.IOException;
 4 
 5 import org.apache.hadoop.conf.Configuration;
 6 import org.apache.hadoop.fs.Path;
 7 import org.apache.hadoop.io.LongWritable;
 8 import org.apache.hadoop.io.Text;
 9 import org.apache.hadoop.mapreduce.Job;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12 
13 
14 /***
15  * 1:用来描述一个特定的作业
16  *       比如,该作业使用哪个类作为逻辑处理中的map,那个作为reduce
17  * 2:还可以指定该作业要处理的数据所在的路径
18  *        还可以指定改作业输出的结果放到哪个路径
19  * @author Administrator
20  *
21  */
22 public class WcRunner {
23 
24     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
25         //创建配置文件
26         Configuration conf = new Configuration();
27         //获取一个作业
28         Job job = Job.getInstance(conf);
29         
30         //设置整个job所用的那些类在哪个jar包
31         job.setJarByClass(WcRunner.class);
32         
33         //本job使用的mapper和reducer的类
34         job.setMapperClass(WcMap.class);
35         job.setReducerClass(WcReduce.class);
36         
37         //指定reduce的输出数据key-value类型
38         job.setOutputKeyClass(Text.class);
39         job.setOutputValueClass(LongWritable.class);
40         
41         
42         //指定mapper的输出数据key-value类型
43         job.setMapOutputKeyClass(Text.class);
44         job.setMapOutputValueClass(LongWritable.class);
45         
46         //指定要处理的输入数据存放路径
47         //FileInputFormat.setInputPaths(job, new Path("hdfs://master:9000/wc/srcdata/"));
48         FileInputFormat.setInputPaths(job, new Path("d:/wc/srcdata/"));
49         
50         
51         //指定处理结果的输出数据存放路径
52         //FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9000/wc/output/"));
53         FileOutputFormat.setOutputPath(job, new Path("d:/wc/output/"));
54         
55         
56         //将job提交给集群运行 
57         job.waitForCompletion(true);
58     }
59     
60 }