1.1MapReduce的作用

假设有一个计算文件中单词个数的需求，文件比较多也比较大，在单击运行的时候机器的内存受限，磁盘受限，运算能力受限，而一旦将单机版程序扩展到集群来分布式运行，将极大增加程序的复杂度和开发难度，因此这个工作可能完成不了。针对以上这个案例，MapReduce在这里能起到什么作用呢，引入MapReduce框架后，开发人员可以将绝大部分工作集中在业务逻辑的开发上，而将分布式计算中的复杂性交由框架来处理。

可见在程序由单机版扩成分布式时，会引入大量的复杂工作。为了提高开发效率，可以将分布式程序中的公共功能封装成框架，让开发人员可以将精力集中于业务逻辑。而MapReduce就是这样一个分布式程序的通用框架。

1.2MapReduce架构图

MapReduce 也采用了 Master/Slave（M/S）架构。它主要由以下几个组件组成 ：Client、JobTracker、 TaskTracker 和 Task。下面分别对这几个组件进行介绍。

（1）Client

用户编写的MapReduce程序通过Client提交到JobTracker端；同时用户可通过Client提供的一些接口查看作业运行状态。在Hadoop内部用“作业” （Job）表示MapReduce程序。一个 MapReduce程序可对应若干个作业，而每个作业会被分解成若干个Map/Reduce任务（Task）。

（2）JobTracker

JobTracker 主要负责资源监控和作业调度。JobTracker 监控所有 TaskTracker 与作业Job的健康状况，一旦发现失败情况后，其会将相应的任务转移到其他节点；同时，JobTracker 会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器，而调度器会在资源出现空闲时，选择合适的任务使用这些资源。在Hadoop 中，任务调度器是一个可插拔的模块，用户可以根据自己的需要设计相应的调度器。

（3）TaskTracker

　　 TaskTracker会周期性地通过Heartbeat将本节点上资源的使用情况和任务的运行进度汇报给JobTracker，同时接收JobTracker发送过来的命令并执行相应的操作（如启动新任务、杀死 任务等）。TaskTracker 使用“slot”等量划分本节点上的资源量。 “slot”代表计算资源（CPU、 内存等）。一个 Task 获取到一个slot 后才有机会运行，而Hadoop调度器的作用就是将各个TaskTracker上的空闲slot分配给Task使用。slot分为Map slot和Reduce slot 两种，分别供Map Task和Reduce Task使用。TaskTracker通过slot数目（可配置参数）限定Task的并发度。

（4）Task

Task 分为 Map Task 和 Reduce Task 两种，均由TaskTracker启动。从上一小节中我们知道，HDFS以固定大小的block 为基本单位存储数据，而对于MapReduce 而言，其处理单位是split。 split 与 block 的对应关系如下图所示。split 是一个逻辑概念，它只包含一些元数据信息，比如 数据起始位置、数据长度、数据所在节点等。它的划分方法完全由用户自己决定。但需要注意的是，split的多少决定了Map Task的数目，因为每个split会交由一个Map Task处理。

Map Task 执行过程如下图所示。由该图可知，Map Task 先将对应的split 迭代解析成一 个个 key/value 对，依次调用用户自定义的map() 函数进行处理，最终将临时结果存放到本地磁盘上，其中临时数据被分成若干个partition(分片)，每个partition 将被一个Reduce Task处理。

Reduce Task 执行过程如下图所示。该过程分为三个阶段：

①从远程节点上读取Map Task 中间结果（称为“Shuffle阶段”）；

②按照key对key/value 对进行排序（称为“Sort阶段”）；

③依次读取 <key, value list>，调用用户自定义的 reduce() 函数处理，并将最终结果存到HDFS上（称为“Reduce 阶段”）。

MapReduce是一种并行编程模式，利用这种模式软件开发者可以轻松地编写出分布式并行程序。在Hadoop的体系结构中，MapReduce是一个简单易用的软件框架，基于它可以将任务分发到由上千台商用机器组成的集群上，并以一种可靠容错的方式并行处理大量的数据集，实现Hadoop的并行任务处理功能。MapReduce框架是由一个单独运行在主节点的JobTrack和运行在每个集群从节点的TaskTrack共同组成的。

主节点负责调度构成一个作业的所有任务，这些任务分布在不同的节点上。主节点监控它们的执行情况，并且重新执行之前失败的任务；

从节点仅负责由主节点指派的任务。

当一个Job任务被提交时，JobTrack接收到提交作业和其配置信息之后，就会配置信息等发给从节点，同时调度任务并监控TaskTrack的执行。

1.3MapReduce程序运行演示

Hadoop的发布包中内置了一个hadoop-mapreduce-example-2.6.5.jar，这个jar包中有各种MR示例程序，可以通过以下步骤运行：

启动hdfs，yarn，然后在集群中的任意一台服务器上启动执行程序（比如运行wordcount）：

hadoop jar hadoop-mapreduce-example-2.6.5.jar wordcount  /wordcount/data /wordcount/out

2.1编程规范

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对象

2.2wordcount 示例编写

需求：在一堆给定的文本文件中统计输出每一个单词出现的总次数

(1)定义一个mapper类

//首先要定义四个泛型的类型
//keyin:  LongWritable    valuein: Text
//keyout: Text            valueout:IntWritablepublic class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{//map方法的生命周期：  框架每传一行数据就被调用一次//key :  这一行的起始点在文件中的偏移量//value: 这一行的内容@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//拿到一行数据转换为stringString line = value.toString();//将这一行切分出各个单词String[] words = line.split(" ");//遍历数组，输出<单词，1>for(String word:words){context.write(new Text(word), new IntWritable(1));}}
}

(2)定义一个reducer类

//生命周期：框架每传递进来一个kv 组，reduce方法被调用一次@Overrideprotected 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集群提交这个jobboolean res = wcjob.waitForCompletion(true);System.exit(res?0:1);}

2.3集群运行模式

1) 将mapreduce程序提交给yarn集群resourcemanager，分发到很多的节点上并发执行

2) 处理的数据和输出结果应该位于hdfs文件系统

3) 提交集群的实现步骤：

将程序打成JAR包，然后在集群的任意一个节点上用hadoop命令启动hadoop jar wordcount.jar cn.bigdata.mrsimple.WordCountDriver inputpath outputpath

作者：杰瑞教育
出处：http://www.cnblogs.com/jerehedu/ 
版权声明：本文版权归烟台杰瑞教育科技有限公司和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。
技术咨询：