| 1. MapReduce 介绍 |
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.MapReduce 编程 |
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: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); }
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/
版权声明：本文版权归烟台杰瑞教育科技有限公司和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。
技术咨询：
