代码分为三个类 WordcountDriver  WordcountMapper   WordcountReduce

package cn.yzx.bigdata.mr.wcemo;import java.io.IOException;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;/** 为啥要main方法：先启动自己，然后把整个程序提交到集群* 相当于yarn集群的客户端，yarn去分配运算资源，然后才能启动* 以为是yarn的客户端，所以要再次封装mr程序的相关运行参数，指定jar包* 最后提交给yarn*/
public class WordcountDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {/** import org.apache.hadoop.mapreduce.xxxx* import org.apache.hadoop.mapred.xxx  mapred是老版本的，我们用MapReduce */Configuration conf = new Configuration();/*conf.set("mapreduce.framework.name", "yarn");conf.set("yarn.resourcemanager.hostname","hadoop01");*///给一些默认的参数Job job = Job.getInstance(conf);//指定本程序的jar包所在的本地路径  把jar包提交到yarnjob.setJarByClass(WordcountDriver.class);/** 告诉框架调用哪个类* 指定本业务job要是用的mapper/Reducer业务类*/job.setMapperClass(WordcountMapper.class);job.setReducerClass(WordcountReduce.class);/** 指定mapper输出数据KV类型*/job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//指定最终的输出数据的kv类型  ，有时候不需要reduce过程，如果有的话最终输出指的就是指reducekv类型  job.setOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);// 指定job的文件输入的原始目录 //paths指你的待处理文件可以在多个目录里边//第一个参数是你给那个job设置  后边的参数 逗号分隔的多个路径 路径是在hdfs里的FileInputFormat.setInputPaths(job, new Path(args[0]));// 指定job 的输出结果所在的目录FileOutputFormat.setOutputPath(job, new Path(args[1]));	/** 找yarn通信* 将job中配置的参数，  以及job所用的java类所在的jar包提交给yarn去运行*//*job.submit();*/// 参数表示程序执行完，告诉我们是否执行成功boolean res = job.waitForCompletion(true);System.exit(res?0:1);}
}

1.2 WordcountMapper 

package cn.yzx.bigdata.mr.wcemo;import java.io.IOException;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/** KEYIN:默认情况下，是mr框架所读到的一行文本的起始偏移量，类型：long * 但是在Hadoop中有更精简的序列化接口，所以不直接用long ，而是用Longwriterable* VALUEIN：默认情况下，是mr框架所读的一行文本的内容， 类型：String  ,同上，用Text(import org.apache.hadoop.io.Text)* * KEYOUT:    是用户自定义逻辑处理完成之后输出数据的key,在此处是单词，类型 String  同上，用Text* VALUEOUT:  是用户自定义逻辑处理完成之后输出数据的value,在此处是单词次数，类型 Integer 同上，用 Intwriterable*//*public class WordcountMapper  extends Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>{}*/
public class WordcountMapper  extends Mapper<LongWritable, Text, Text, IntWritable>{/** map 阶段的业务逻辑就写在自定义的map   方法中* maptask会对每一行数据调用一次我们自定义的map方法* */@Overrideprotected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {//将maptask传给我们的内容先转换成StringString line= value.toString();//根据空格这一行切成单词String[] words = line.split(" +");//将单词输出为<单词，1>for(String word:words) {//将单词作为key,将次数1作为value，以便后续的数据分发，可以根据单词分发，以便相同的单词会到相同reduce taskcontext.write(new Text(word),new IntWritable(1));}}}

package cn.yzx.bigdata.mr.wcemo;import java.io.IOException;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;/** KEYIN, VALUEIN, 对应mapper的KEYOUT,VALUEOUT  类型对应* * KEYOUT, VALUEOUT,是自定义reduce逻辑处理结果的输出数据类型* * KEYOUT是单词* VALUEOUT是总次数*/
/*public class WordcountReduce extends Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>{}*/
public class WordcountReduce extends Reducer<Text, IntWritable, Text, IntWritable>{/** 入参key,是一组相同单词kv对的key* <dayu,1><dayu,1><dayu,1><dayu,2>* <hello,1><hello,1><hello,1><hello,1>* <world,2><world,2><world,2><world,2>* 上面三个分为三组，每组都调用一次reduce方法*/@Overrideprotected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {int count=0;for(IntWritable value:values) {count+=value.get();}//每组的统计结果context.write(key, new IntWritable(count));}
}

2.执行

打包发到Linux 上，

执行 stop-all.sh 

hadoop jar wordcount.jar cn.yzx.bigdata.mr.wcemo.WordcountDriver    /wordcount/input /wordcount/output/output

（1）用户编写的程序分成三个部分：Mapper，Reducer，Driver(提交运行mr程序的客户端)

（2）Mapper的输入数据是KV对的形式（KV的类型可自定义）

（3）Mapper的输出数据是KV对的形式（KV的类型可自定义）

（4）Mapper中的业务逻辑写在map()方法中

（5）map()方法（maptask进程）对每一个<K,V>调用一次

（6）Reducer的输入数据类型对应Mapper的输出数据类型，也是KV

（7）Reducer的业务逻辑写在reduce()方法中

（8）Reducetask进程对每一组相同k的<k,v>组调用一次reduce()方法

（9）用户自定义的Mapper和Reducer都要继承各自的父类

（10）整个程序需要一个Drvier来进行提交，提交的是一个描述了各种必要信息的job对象