内容
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函数对该键值对进行计算,输出若干键值对。<key, value, partition>
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个<hello,1>压缩为1个<hello, 10w>通过网络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:
<hello, 1> 1000万个 <hello, 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接口。
作业:
- mapreduce处理过程,自己的语言写
- 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
访问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 </path/to/your/jar.jar> /<inputpath> /<outputpath>