MapReduce概述及工作流程

内容

mapreduce原语（独创）

mapreduce工作流程（重点）

MR作业提交流程（重点）

YARN RM-HA搭建（熟练）

运行自带的wordcount（了解）

动手写wordcount（熟练）

MapReduce原语

hadoop MapReduce框架可以让你的应用在集群中

可靠地

容错地

并行

处理TB级别的数据

1024TB=1PB  1024PB=1EB  1024EB=1ZB

MapReduce原语

 

“相同”key的键值对为一组调用一次reduce方法，方法内迭代这一组数据进行计算

分组比较器

YARN:资源管理框架

         ResourceManager：一个   主

         NodeManager：很多，每个DataNode上有一个  从

         Container（容器）：CPU、内存

公司为了营业，挣钱租老王家的写字楼

公司相当于MR作业

MR任务相当于公司员工，员工干活，相当于MR的任务运行。

员工在办公室干活，任务在容器运行。

每个容器同时运行一个任务

         客人提出订几间房

 

1、一个ResourceManager主节点

2、每个DataNode上一个NodeManager从节点

3、每个运行于MapReduce的程序有一个MRAppMaster

公司的运作流程

 

1、MapReduce将输入的数据集逻辑切片 split

2、map任务以并行方式处理切片数据

3、框架对map输出排序，然后将数据发送给reduce

4、MapReduce的输入输出数据存在于同一个文件系统（HDFS）

5、框架负责任务调度、任务监控和失败任务的重新执行

容错地、可靠地、并行计算

1、MapReduce处理键值对形式的很多键值对输入，生成键值对形式的很多键值对输出

2、框架会对键和值序列化，因此键类型和值类型需要实现Writable接口。框架会对键进行排序，因此必须实现WritableComparable接口。

3、map输出键值对类型和reduce键值对输入类型一致

4、map的输入键值对类型和输出键值对类型一般不一致

5、reduce的输入键值对类型和输出键值对类型一般不一致

尽管hadoop框架是java开发的，MapReduce应用不一定得java开发。

hadoop streaming允许用户使用可执行文件的方式提供mapper和reducer，创建和执行作业。

hadoop pipes是一个跟SWIG兼容的C++ API，用于开发MapReduce应用（不基于JNI）。

mapreduce工作流程

为什么叫MapReduce：MapTask & ReduceTask

1、每个block会有map任务

2、block切分为切片，每个切片对应一个map任务，默认一个block一个切片，一个map

3、map默认按行读取切片数据，组成键值对<当前行字节偏移量, "读到的行字符串">

4、map函数对该键值对进行计算，输出若干键值对。

         partition指定该键值对由哪个reducer进行处理

5、map输出的kvp写到环形缓冲区，环形缓冲区默认100MB，阈值80%，当环缓达到80%就向磁盘溢写小文件，该小文件首先按照分区号排序，相同分区号的按key进行排序。

6、默认如果落磁盘的小文件达到了3个，则进行归并，归并的大文件也是按分区号排序，相同分区号按照key进行排序。只是一个归并。

7、如果map任务处理完了，它的输出被下载到reducer所在主机

         按照HTTP GET的方式下载到reducer：

         reducer发送HTTP GET请求到mapper主机下载数据，该过程是洗牌shuffle

8、每个map任务都要经历运行结束洗牌的过程

9、可以设置combinClass，先在map端对数据进行一个压缩，比如10w个压缩为1个通过网络IO洗牌，肯定要快很多。一般情况下，combineClass就是一个reducerClass。

combinerClass的设置要求数据算法满足结合律。

交换律

         1+2=2+1

结合律    

1+2+3=(1+2)+3=1+(2+3)

        

map1   5/3

map2   7/6       reduce:  5/3+7/6+8/11 =?  reduce:（5+7+8）/(3+6+11)

map3   8/11

map任务结束

reeduce任务开始

9、等所有map任务都运行结束，并且洗牌结束，每个reducer获取到它自己应得的所有数据，此时开始reducer处理过程。

10、如果有时间，reduce会对洗牌获取的数据进行归并落磁盘

         如果没有时间，也归并，只是可能不落磁盘，直接交给reduce方法进行迭代处理了。

         洗牌获取到的数据也可能不落磁盘，此时归并的键值对来源可能是磁盘的和内存的一个混合。

11、reduce按照key进行分组，每个分组调用一次reduce方法，该方法迭代计算，将结果写到HDFS输出。

当一个map任务计算结束，所有的reduce需要使用http get请求获取各自分区编号的数据，当所有map任务结束后，开始reduce计算阶段。

blk按照设置进行切片，一个切片对应一个map任务，map按行读取切片内容，以键值对的形式发给map方法（<"偏移量"， “zifuchuan”>）

当map对当前简直对计算完成，要写到环形缓冲区，在写之前要计算该键值对的分区编号

默认情况下，key的hash值对reduce个数取模。

当环形缓冲区大小达到到80%的时候，需要向磁盘溢写数据，在溢写的时候需要对键值对按照分区排序，分区内按照key的字典序排序（快排排序）

溢写的小文件如果达到3个，则进行归并，归并为大文件，大文件也是按照分区排序，分区内按照key的字典序排序。

当一个map任务处理完它的切片的数据，此时所有的reduce任务到该map的机器以http get请求获取各自编号分区的数据，下载到reduce本地

reduce获取到map的数据后，如果有时间，也会进行归并

并不能保证此时所有的map都计算结束了。

只有当所有的map计算结束，同时reduce获取到所有的数据之后，才开始进行reduce计算。

按照原语，相同key的键值对为一组，调用一次reduce方法，方法内迭代这组数据计算，结果输出到HDFS中。

mapreduce是一套分布式计算的流程、框架

数单词游戏：

 

 

getFileBlockLocations(new Path(), offset, len);

 

reduce从map端拉取数据的过程称为洗牌shuffle

通过网络拉取，慢！！！

要对map端数据进行压缩：

Combiner：

  1000万个     

但是不能保证combiner什么时候都能用：

需要计算满足结合律：（A+B）+C=A+(B+C)

job.setCombinerClass(MyReducer.class)

8/9

4/7       REDUCE: (8+4+2)/(9+7+11)

2/11

也不能保证combiner什么时候都用得上：

环形缓冲区小文件归并，进行combiner，如果不归并，没有combiner过程。

reducer通过HTTP按照分区号获取map输出文件的数据。map端有一个HTTP服务处理该reducer的HTTP请求。该HTTP服务最大线程数由mapreduce.shuffle.Max.threads属性指定。这个属性指定nodemanager的线程数，而不是对map任务指定线程数（该数字在多个不同的任务之间共享），因为nodemanager上有可能运行了好几个map任务。默认值是0，表示最大线程数是服务器处理器核心数的两倍。

map输出文件位于运行map任务的本地磁盘。一个reduce任务需要从集群中多个map任务获取指定分区的数据。多个map任务有可能是在不同时间完成的，每当一个map任务运行完，reduce就从该map任务获取指定分区数据。reduce任务会以多线程的方式从多个map任务并行获取指定分区数据。默认线程数是5，可以通过mapreduce.reduce.shuffle.parallelcopies属性指定。

reducer拷贝map的输出如果很小，则放在内存中（mapreduce.reduce.shuffle.input.buffer.percent指定堆空间百分比）否则拷贝到磁盘。当内存缓冲区数据大小达到阈值（mapreduce.reduce.shuffle.merge.percent

）或map输出文件个数达到阈值（mapreduce.reduce.merge.inmem.threshold

），就发生文件合并溢写到磁盘上。如果指定combiner，此处也会进行combine。

二次排序（先了解）

在map阶段按照key对键值对进行排序，对值不排序。如果相对value进行排序，就需要二次排序。

需求：查找每年的最高气温

数据格式：年份为key，每天的气温是value

所谓二次排序：

1、新的key应该是输入的key和value的组合

2、按照复合key进行比较排序

3、分区比较器和分组比较器只对复合key中的原生key进行分区和分组

总结

Map：

1、根据业务需求处理数据并映射为KV模型

2、并行分布式

3、计算向数据移动

Reduce：

1、数据全量/分量加工

2、Reducer中可以包含不同的key     分区的范围大于分组

3、相同分区的Key汇聚到一个Reducer中

4、“相同”的Key调用一次reduce方法

5、排序和比较实现key的汇聚

K,V使用自定义数据类型           MyKey：WritableComparable

MyValue：Writable

1、节省开发成本，提高程序自由度

2、框架会对键和值序列化，因此键类型和值类型需要实现Writable接口。

3、框架会对键进行排序，因此必须实现WritableComparable接口。

作业：

1. mapreduce处理过程，自己的语言写
2. java API操作HDFS

MR作业提交流程

YARN

ResourceManager管理集群中所有的资源

通过NodeManager管理

NodeManager通过Container管理资源

         Container包装资源:CPU/内存/IO

MapReduce作业

AppMaster  调度

         向RM申请资源

MapTask

ReduceTask

客户端：

         RM客户端：用于申请资源

         AM客户端：用于跟AppMaster交互

YARN：解耦资源与计算

ResourceManager

主，核心

集群节点资源管理

NodeManager

与RM汇报资源

管理Container生命周期

计算框架中的资源都以Container表示

Container：【由节点NM管理，CPU,MEM,I/O大小，启动命令】

         内存：1024MB

         CPU：1个虚拟核心 vcore

默认NodeManager启动线程监控Container大小，超出申请资源额度，kill

支持Linux内核的Cgroup

MR ：

AppMaster  拥有 RM客户端

作业为单位，避免单点故障，负载到不同的节点

创建Task，需要和RM申请资源（Container）

Task-Container

         Map任务

         Reduce任务

Client：

RM-Client：请求资源创建AM

AM-Client：与AM交互

YARN：Yet Another Resource Negotiator；

Hadoop 2.0新引入的资源管理系统，直接从MRv1演化而来的；

核心思想：将MRv1中JobTracker的资源管理和任务调度两个功能分开，分别由ResourceManager和ApplicationMaster进程实现

ResourceManager：负责整个集群的资源管理和调度

ApplicationMaster：负责应用程序相关的事务，比如任务调度、任务监控和容错等

YARN的引入，使得多个计算框架可运行在一个集群中

每个应用程序对应一个ApplicationMaster

目前多个计算框架可以运行在YARN上，比如MapReduce、Spark、Storm等

MapReduce On YARN：MRv2

将MapReduce作业直接运行在YARN上，而不是由JobTracker和TaskTracker构建的MRv1系统中

基本功能模块

YARN：负责资源管理和调度

MRAppMaster：负责任务切分、任务调度、任务监控和容错等

MapTask/ReduceTask：任务驱动引擎，与MRv1一致

每个MapRduce作业对应一个MRAppMaster

MRAppMaster任务调度

YARN将资源分配给MRAppMaster

MRAppMaster进一步将资源分配给内部的任务

MRAppMaster容错

失败后，由YARN重新启动

任务失败后，MRAppMaster重新申请资源

ResourceManager挂怎么办？RM-HA

流程

1、客户端，提交MapReduce作业

2、YARN的资源管理器（Resource Manager），协调集群中计算资源的分配

3、YARN的节点管理器（Node Manager），启动并监控集群中的计算容器

4、MapReduce的Application Master，协调MapReduce作业中任务的运行。Application Master和MapReduce任务运行于容器中，这些容器由resourcemanager调度，由nodemanager管理。

5、分布式文件系统（一般是HDFS），在组件之间共享作业数据。

Job对象的submit方法创建了一个内部的JobSubmitter实例并调用该实例的submitJobInternal方法。一旦提交了作业，waitForCompletion方法每秒钟轮询作业的执行进度，如果进度发生了变化，则向控制台报告进度。当作业成功完成，展示作业计数器的数据。否则展示作业失败的错误日志信息。

客户端：JobSubmitter实现的作业提交的过程有如下几个步骤：

1、向resourcemanager申请一个新的application ID，用于MapReduce作业的ID

2、检查作业的输出。如果没有指定输出或者输出路径已经存在，则不提交作业，MapReduce程序抛异常

3、计算作业的输入切片。如果不能计算切片（比如输入路径不存在等），不提交作业，MR程序抛异常。

4、拷贝执行作业需要的资源到共享文件系统的以作业ID命名的目录中，这些资源包括作业的jar包，配置文件，计算好的输入切片。作业的jar包有一个很高的副本数量（mapreduce.client.submit.file.replication指定，默认值是10），这样当nodemanager如果运行作业中的任务，会有很多副本可以访问。

5、调用resourcemanager的submitApplication方法提交作业。

1、YARN为请求分配一个容器，resourcemanager通过容器所在节点上的nodemanager在该容器中启动application master进程。

2、MapReduce作业的application master是一个java app，主入口类是MRAppMaster。从HDFS抽取客户端计算好的输入切片，为每一个切片创建一个map任务对象，以及一定数量的reduce任务对象.

application master会为作业中所有的map任务以及reduce任务向resourcemanager请求容器。为map任务的请求会首先进行并且相对于reduce任务请求有更高的优先级。当map任务完成率达到了5%之后才会为reduce任务发送容器请求。

appmaster从hdfs抽取客户端上传的信息，计算好map对象和reduce对象，首先向resourcemanager为map任务申请资源，当map任务完成5%之后为reduce任务申请资源

reduce任务可以运行于集群中的任意位置，而map任务会有本地读取数据的限制。移动计算而不是数据。数据本地。次之为机架本地。

请求会指定每个任务需要的内存和cpu资源。默认情况下为每个map任务或reduce任务分配1024MB的内存和一个虚拟核心。这些值对于每个作业都是可以配置的：mapreduce.map.memory.mb，

mapreduce.reduce.memory.mb

mapreduce.map.cpu.vcores

以及mapreduce.reduce.cpu.vcores。

一旦resourcemanager在一个节点上的一个容器中为一个任务分配了资源，application master与nodemanager通信，启动容器。任务通过一个java app来执行，该app的主入口类是YarnChild。在它可以开始任务的执行之前，它要本地化任务需要的资源，包括jar包，配置文件，以及分布式缓存中存储的其他共享文件。最后，它开始运行map任务或者reduce任务。

当作业的最后一个任务完成并通知application master，AppMaster就更改作业的状态为”successfully”。作业就打印信息告知客户端，客户端waitForCompletion方法返回。此时也会在控制台打印作业的统计信息和计数器的信息。

作业完成，application master所在容器和任务所在容器销毁工作状态（中间的输出结果删除）。作业的信息被作业历史服务器存档以备以后查询使用。

YARN RM-HA搭建

mapred-site.xml

local/classic/yarn

指定mr作业运行的框架：要么本地运行，要么使用MRv1，要么使用yarn

yarn-site.xml

将配置文件在四台服务器同步

scp   node[234]:`pwd`

首先启动HDFS

start-ha.sh

在node3或node4上执行命令：

start-yarn.sh

在node4或者node3上执行命令：

yarn-daemon.sh  start  resourcemanager

停止：

         在node3或者node4上执行：

         stop-yarn.sh

         在node4或者node3上执行：

         yarn-deamon.sh  stop  resourcemanager

http://node3:8088

http://node4:8088

访问resourcemanager的web页面

运行自带的wordcount

运行的命令：

cd $HADOOP_HOME

cd share/hadoop/mapreduce

hadoop jar hadoop-mapreduce-examples-2.6.5.jar wordcount  /input  /output

*input:是hdfs文件系统中数据所在的目录

*ouput:是hdfs中不存在的目录，mr程序运行的结果会输出到该目录

输出目录内容：

-rw-r--r--   3 root supergroup          0 2017-07-02 02:49 /mr/test/output/_SUCCESS

-rw-r--r--   3 root supergroup         49 2017-07-02 02:49 /mr/test/output/part-r-00000

/_SUCCESS：是信号/标志文件

/part-r-00000：是reduce输出的数据文件

r：reduce的意思，00000是对应的reduce编号，多个reduce会有多个数据文件

启动脚本和停止脚本：

start-hdfs-ha-rm-ha.sh

stop-hdfs-ha-rm-ha.sh

动手写wordcount

1、新建eclipse的java项目

2、添加hadoop的jar包依赖

121个jar包

$HADOOP_HOME/share/hadoop/{common,common/lib,hdfs,hdfs/lib,mapreduce,mapreduce/lib,tools/lib,yarn,yarn/lib}.jar

3、添加hadoop的配置文件到类路径

从集群拷贝这四个文件到当前项目类路径

core-site.xml

hdfs-site.xml

mapred-site.xml

yarn-site.xml

4、编写Mapper、Reducer以及MainClass

wordcount

WCMapper.java

WCReducer.java

MainClass.java

5、打包

只打包三个类就可以。

6、上传

7、运行

yarn   jar     /  /