编程规范
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对象。
wordcount示例编写
需求：在一堆给定的文本文件中统计输出每一个单词出现的总次数。
(1)定义一个mapper类
//首先要定义四个泛型的类型 //keyin: LongWritable valuein: Text //keyout: Text valueout:IntWritable public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ //map方法的生命周期： 框架每传一行数据就被调用一次 //key : 这一行的起始点在文件中的偏移量 //value: 这一行的内容 @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //拿到一行数据转换为string String line = value.toString(); //将这一行切分出各个单词 String[] words = line.split(" "); //遍历数组，输出<单词，1> for(String word:words){ context.write(new Text(word), new IntWritable(1)); } } }
(2)定义一个reducer类
//生命周期：框架每传递进来一个kv 组，reduce方法被调用一次 @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { //定义一个计数器 int count = 0; //遍历这一组kv的所有v，累加到count中 for(IntWritable value:values){ count += value.get(); } context.write(key, new IntWritable(count)); } }
(3)定义一个主类，用来描述job并提交job
public class WordCountRunner { //把业务逻辑相关的信息（哪个是mapper，哪个是reducer，要处理的数据在哪里，输出的结果放哪里……）描述成一个job对象 //把这个描述好的job提交给集群去运行 public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job wcjob = Job.getInstance(conf); //指定我这个job所在的jar包 // wcjob.setJar("/home/hadoop/wordcount.jar"); wcjob.setJarByClass(WordCountRunner.class); wcjob.setMapperClass(WordCountMapper.class); wcjob.setReducerClass(WordCountReducer.class); //设置我们的业务逻辑Mapper类的输出key和value的数据类型 wcjob.setMapOutputKeyClass(Text.class); wcjob.setMapOutputValueClass(IntWritable.class); //设置我们的业务逻辑Reducer类的输出key和value的数据类型 wcjob.setOutputKeyClass(Text.class); wcjob.setOutputValueClass(IntWritable.class); //指定要处理的数据所在的位置 FileInputFormat.setInputPaths(wcjob, "hdfs://hdp-server01:9000/wordcount/data/big.txt"); //指定处理完成之后的结果所保存的位置 FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://hdp-server01:9000/wordcount/output/")); //向yarn集群提交这个job boolean res = wcjob.waitForCompletion(true); System.exit(res?0:1); }
