04 MapReduce原理介绍

大数据实战（上）

# MapReduce原理介绍

大纲：

* Mapreduce介绍

* MapReduce2运行原理

* shuffle及排序

 

 定义

* Mapreduce 最先是由google公司研究提出的一种免息nag大规模数据处理的并行计算模型和方法。是hadoop面向大数据并行处理的计算模型、框架和平台

* Mapreduce是一个计算框架，既然是作计算的框架，那么表现形式就是有个输入（input），mapreduce操做这个输入（input），经过自己定义好的计算模型，获得一个输出（output），这个输出就是咱们所须要的结果。

 

 第一部分：MapReduce工做原理

 

 
MapReduce 角色
•Client ：做业提交发起者。
•JobTracker: 初始化做业，分配做业，与TaskTracker通讯，协调整个做业。
•TaskTracker：保持JobTracker通讯，在分配的数据片断上执行MapReduce任务。
提交做业
•在做业提交以前，须要对做业进行配置
•程序代码，主要是本身书写的MapReduce程序。
•输入输出路径
•其余配置，如输出压缩等。
•配置完成后，经过JobClinet来提交
做业的初始化
•客户端提交完成后，JobTracker会将做业加入队列，而后进行调度，默认的调度方法是FIFO调试方式。
任务的分配
•TaskTracker和JobTracker之间的通讯与任务的分配是经过心跳机制完成的。
•TaskTracker会主动向JobTracker询问是否有做业要作，若是本身能够作，那么就会申请到做业任务，这个任务可使Map也多是Reduce任务。
任务的执行
•申请到任务后，TaskTracker会作以下事情：
  •拷贝代码到本地
  •拷贝任务的信息到本地
  •启动JVM运行任务
状态与任务的更新
•任务在运行过程当中，首先会将本身的状态汇报给TaskTracker，而后由TaskTracker汇总告之JobTracker。
•任务进度是经过计数器来实现的。


做业的完成
•JobTracker是在接受到最后一个任务运行完成后，才会将任务标志为成功。
•此时会作删除中间结果等善后处理工做。



第二部分：错误处理
任务失败

•MapReduce在设计之出，就假象任务会失败，因此作了不少工做，来保证容错。
•一种状况:    子任务失败
•另外一种状况：子任务的JVM忽然退出
•任务的挂起
TaskTracker失败
•TaskTracker崩溃后会中止向Jobtracker发送心跳信息。
•Jobtracker会将该TaskTracker从等待的任务池中移除。并将该TaskTracker上的任务，移动到其余地方去从新运行。
•TaskTracker能够被JobTracker放入到黑名单，即便它没有失败。


JobTracker失败
•单点故障，Hadoop新的0.23版本解决了这个问题。
第三部分：做业调度
FIFO
         Hadoop 中默认的调度器，它先按照做业的优先级高低，再按照到达时间的前后选 择被执行的做业

公平调度器
      为任务分配资源的方法，其目的是随着时间的推移，让提交的做业获取等量的集 群共享资源，让用户公平地共享集群。具体作法是：当集群上只有一个任务在运行 时，它将使用整个集群，当有其余做业提交时，系统会将TaskTracker节点空间的时 间片分配给这些新的做业，并保证每一个任务都获得大概等量的CPU时间。

容量调度器

             支持多个队列，每一个队列可配置必定的资源量，每一个队列采用 FIFO 调度策略，为 了防止同一个用户的做业独占队列中的资源，该调度器会对同一用户提交 的做业所 占资源量进行限定。调度时，首先按如下策略选择一个合适队列：计算每一个队列中正在运行的任务数与其应该分得的计算资源之间的比值，选择一个该比值 最小的队 列；而后按如下策略选择该队列中一个做业：按照做业优先级和提交时间顺序选择 ，同时考虑用户资源量限制和内存限制。可是不可剥夺式
配置公平调度器
1.修改mapred-stie.xml 加入以下内容

<property>
      <name>mapred.jobtracker.taskScheduler</name>
      <value>org.apache.hadoop.mapred.FairScheduler</value>
       </property>
       <property>
      <name>mapred.fairscheduler.allocation.file</name>
      <value>/opt/hadoop/conf/allocations.xml</value>
       </property>
        <property>
     <name>mapred.fairscheduler.poolnameproperty</name>
      <value>pool.name</value>
       </property>

2 . 在 Hadoop conf 下建立
    allocations.xml
   内容为
    

 <?xml version="1.0"?>
        <alloctions>
        </alloctions>
样例：
<pool name="sample_pool">
<minMaps>5</minMaps>
<minReduces>5</minReduces>
<weight>2.0</weight>
</pool>
<user name="sample_user">
<maxRunningJobs>6</maxRunningJobs>
</user>
<userMaxJobsDefault>3</userMaxJobsDefault>

　　

3. 重启 JobTracker
4. 访问 http://jobTracker:50030/scheduler , 查看 FariScheduler 的 UI
5 . 提交任务测试




第四部分：Shuffle与排序

 

 

          Mapreduce 的 map 结束后，把数据从新组织，做为 reduce 阶段的输入，该过程称 之为 shuffle--- 洗牌。
          而数据在 Map 与 Reduce 端都会作排序。
Map

• Map 的输出是由collector控制的
• 咱们从collect函数入手
Reduce
•reduce的Shuffle过程，分红三个阶段：复制Map输出、排序合并、reduce处理。
•主要代码在reduce的 run函数

Shuffle优化

•首先Hadoop的Shuffle在某些状况并非最优的，例如，若是须要对2集合合并，那么其实排序操做时不须要的。
•咱们能够经过调整参数来优化Shuffle
•Map端
•io.sort.mb
•Reduce端
•mapred.job.reduce.input.buffer.percent

第五部分：任务的执行时的一些特有的概念
推测式执行
•每一道做业的任务都有运行时间，而因为机器的异构性，可能会会形成某些任务会比全部任务的平均运行时间要慢不少。
•这时MapReduce会尝试在其余机器上重启慢的任务。为了是任务快速运行完成。
•该属性默认是启用的。


JVM重用

•启动JVM是一个比较耗时的工做，因此在MapReduce中有JVM重用的机制。
•条件是统一个做业的任务。
•能够经过mapred.job.reuse.jvm.num.tasks定义重用次数，若是属性是-1那么为无限制。




跳过坏记录
•数据的一些记录不符合规范，处理时抛出异常，MapReduce能够讲次记录标为坏记录。重启任务时会跳过该记录。
•默认状况下该属性是关闭的。

任务执行环境
•Hadoop为Map与Reduce任务提供运行环境。
•如：Map能够知道本身的处理的文件
•问题：多个任务可能会同时写一个文件
•解决办法：将输出写到任务的临时文件夹。目录为：{mapred.out. put.dir}/temp/${mapred.task.id}





第六部分：MapReduce的类型与格式
类型
•MapReduce的类型 使用键值对做为输入类型(key,value)•输入输出的数据类型是经过输入输出的格式进行设定的。
输入格式
•输入分片与记录
•文件输入
•文本输入
•二进制输入
•多文件输入
•数据库格式的输入

输入分片与记录
•Hadoop经过InputSplit表示分片。
•一个分片并非数据自己，而是对分片数据的引用。
•InputFormat接口负责生成分片

<ignore_js_op> 

文件输入
•实现类：FileInputFormat
•经过文件做为输入源的基类。
•四个方法：
•addInputPath()
•addInputPaths()
•setInputPath()
•setInputPaths()
•FileInputFormat会按HDFS块的大小来分割文件
•避免分割
•继承FileInputFormat 重载isSplitable()
•return false

文本输入

•实现类：TextInputFormat
•TextInputFormat 是默认的输入格式。
•包括：
•KeyValueTextInputFormat
•NLineInputFormat
•XML
•输入分片与HDFS块之间的关系
•TextInputFormat的某一条记录可能跨块存在

二进制输入

•实现类：SequenceFileInputFormat
•处理二进制数据
•包括：
•SequenceFileAsTextInputFormat
•SequenceFileAsBinaryInputFormat

多文件输入

•实现类：MultipleInputs
•处理多种文件输入
•包括：
•addInputPath

数据库输入

•实现类：DBInputFormat
•注意使用，由于链接过多，数据库没法承受。

输出格式
•文本输出
•二进制输出
•多文件输出
•数据库格式的输出
文本输出
•实现类：TextOutputFormat
•默认的输出方式
• 以 "key \t value" 的方式输出
二进制输出

•基类： SequenceFileOutputFormat
•实现类： SequenceFileAsTextOutputFormat
              MapFileOutputFormat
              SequenceFileAsBinaryOutputFormat

多文件输出

•MutipleOutputFormat•MutipleOutputs
•二者的不一样在于MutipleOutputs能够产生不一样类型的输出
数据库格式输出• 实现类 DBOutputFormat