一脸懵逼学习MapReduce的原理和编程（Map局部处理，Reduce汇总）和MapReduce几种运行方式

1：MapReduce的概述：

　　（1）：MapReduce是一种分布式计算模型，由Google提出，主要用于搜索领域，解决海量数据的计算问题.
　　（2）：MapReduce由两个阶段组成：Map和Reduce，用户只需要实现map()和reduce()两个函数，即可实现分布式计算，非常简单。
　　（3）：这两个函数的形参是key、value对，表示函数的输入信息。

2：MapReduce执行步骤：

　　（1）： map任务处理

　　　　（a）:读取输入文件内容，解析成key、value对。对输入文件的每一行，解析成key、value对。每一个键值对调用一次map函数。
　　　　（b）:写自己的逻辑，对输入的key、value处理，转换成新的key、value输出。
　　（2）reduce任务处理

　　　　（a）在reduce之前，有一个shuffle的过程对多个map任务的输出进行合并、排序。
　　　　（b）写reduce函数自己的逻辑，对输入的key、value处理，转换成新的key、value输出。
　　　　（c）把reduce的输出保存到文件中。
　　　　　　例子：实现WordCountApp
3：map、reduce键值对格式：
一脸懵逼学习MapReduce的原理和编程（Map局部处理，Reduce汇总）和MapReduce几种运行方式

4：MapReduce流程：
　　（1）代码编写
　　（2）作业配置
　　（3）提交作业
　　（4）初始化作业
　　（5）分配任务
　　（6）执行任务
　　（7）更新任务和状态
　　（8）完成作业

5：MapReduce介绍及wordcount和wordcount的编写和提交集群运行的案例：

WcMap类进行单词的局部处理：

 package com.mapreduce;

 import java.io.IOException;

 import org.apache.commons.lang.StringUtils;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Mapper;

 /***

  *

  * @author Administrator

  * 1：4个泛型中，前两个是指定mapper输入数据的类型，KEYIN是输入的key的类型，VALUEIN是输入的value的值

  *       KEYOUT是输入的key的类型，VALUEOUT是输入的value的值

  * 2：map和reduce的数据输入和输出都是以key-value的形式封装的。

  * 3：默认情况下，框架传递给我们的mapper的输入数据中，key是要处理的文本中一行的起始偏移量，这一行的内容作为value

  * 4：key-value数据是在网络中进行传递，节点和节点之间互相传递，在网络之间传输就需要序列化，但是jdk自己的序列化很冗余

  *    所以使用hadoop自己封装的数据类型，而不要使用jdk自己封装的数据类型；

  *    Long--->LongWritable

  *    String--->Text

  */

 public class WcMap extends Mapper<LongWritable, Text, Text, LongWritable>{

     //重写map这个方法

     //mapreduce框架每读一行数据就调用一次该方法

     @Override

     protected void map(LongWritable key, Text value, Context context)

             throws IOException, InterruptedException {

         //具体业务逻辑就写在这个方法体中，而且我们业务要处理的数据已经被框架传递进来，在方法的参数中key-value

         //key是这一行数据的起始偏移量，value是这一行的文本内容

         //1:切分单词,首先拿到单词value的值,转化为String类型的

         String str = value.toString();

         //2:切分单词,空格隔开,返回切分开的单词

         String[] words = StringUtils.split(str," ");

         //3:遍历这个单词数组，输出为key-value的格式，将单词发送给reduce

         for(String word : words){

             //输出的key是Text类型的，value是LongWritable类型的

             context.write(new Text(word), new LongWritable());

         }

     }

 }

WcReduce进行单词的计数处理：

 package com.mapreduce;

 import java.io.IOException;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Reducer;

 /***

  *

  * @author Administrator

  * 1:reduce的四个参数，第一个key-value是map的输出作为reduce的输入,第二个key-value是输出单词和次数，所以

  *      是Text,LongWritable的格式；

  */

 public class WcReduce extends Reducer<Text, LongWritable, Text, LongWritable>{

     //继承Reducer之后重写reduce方法

     //第一个参数是key，第二个参数是集合。

     //框架在map处理完成之后，将所有key-value对缓存起来，进行分组，然后传递一个组<key,valus{}>，调用一次reduce方法

     //<hello,{1,1,1,1,1,1.....}>

     @Override

     protected void reduce(Text key, Iterable<LongWritable> values,Context context)

             throws IOException, InterruptedException {

         //将values进行累加操作，进行计数

         long count = ;

         //遍历value的list，进行累加求和

         for(LongWritable value : values){

             count += value.get();

         }

         //输出这一个单词的统计结果

         //输出放到hdfs的某一个目录上面，输入也是在hdfs的某一个目录

         context.write(key, new LongWritable(count));

     }

 }

WcRunner用来描述一个特定的作业

 package com.mapreduce;

 import java.io.IOException;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.LongWritable;

 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;

 /***

  * 1:用来描述一个特定的作业

  *       比如,该作业使用哪个类作为逻辑处理中的map,那个作为reduce

  * 2:还可以指定该作业要处理的数据所在的路径

  *        还可以指定改作业输出的结果放到哪个路径

  * @author Administrator

  *

  */

 public class WcRunner {

     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

         //创建配置文件

         Configuration conf = new Configuration();

         //获取一个作业

         Job job = Job.getInstance(conf);

         //设置整个job所用的那些类在哪个jar包

         job.setJarByClass(WcRunner.class);

         //本job使用的mapper和reducer的类

         job.setMapperClass(WcMap.class);

         job.setReducerClass(WcReduce.class);

         //指定reduce的输出数据key-value类型

         job.setOutputKeyClass(Text.class);

         job.setOutputValueClass(LongWritable.class);

         //指定mapper的输出数据key-value类型

         job.setMapOutputKeyClass(Text.class);

         job.setMapOutputValueClass(LongWritable.class);

         //指定要处理的输入数据存放路径

         FileInputFormat.setInputPaths(job, new Path("hdfs://master:9000/wc/srcdata"));

         //指定处理结果的输出数据存放路径

         FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9000/wc/output"));

         //将job提交给集群运行

         job.waitForCompletion(true);

     }

 }

书写好上面的三个类以后打成jar包上传到虚拟机上面进行运行：
一脸懵逼学习MapReduce的原理和编程（Map局部处理，Reduce汇总）和MapReduce几种运行方式