MapReduce架构

Mapreducer分为两个阶段：Map映射、Reduce归约
Map：对数据进行处理
Reduce：对Map结果进行汇总
Shuffle：‘map‘方法之后,‘reducer‘方法之前
AppMaster：负责整个MR的过程调度及状态协调
MapTask：负责Map阶段的整个数据处理流程
ReduceTask：负责Reduce阶段的整个数据处理流程

MapReduce阶段处理

              map()       reduce()
   Input ---> Mapper ---> Reducer ---> Output
   	  |	       |           |  
     InputFormat    Shuffle    OutpuFormat
          |                        |  
    FileInputFormat          FileOutputFormat
          |                        |
    TextInputFormat          TextOutputFormat
  	  |  	RecordReader       |   	 RecordReader
         ↓          |             ↓          | 
         new  LineRecordReader    new  LineRecordReader

Inputformat中定义了两个抽象方法: getSplits() 、 creatRecordReader()
FileInputFormat 继承 Inputformat 实现了其中一个抽象方法：getSplits(),对Input中传进来的文件进行切片。
TextInputFormat 继承 FileInputFormat 实现了另一个抽象方法：creatRecordReader(),还继承了isSplitable()。
creatRecordReader()方法创建了一个 LineRecordReader 对象,对切片后的数据转换成KV键值对的形式传输给Mapper，就是这个对象决定了 Mapper的输入数据类型,KeyLongWritable、Value是Text。

InputFormat：

1.客户端提交任务前先获取待处理文件的信息,然后通过 Connect() 判断文件是提交到本地集群还是远程集群的 yarn上,本地走localRunner,远程则走YarnRunner。
2.通过 CheckSpace(job)检查输出路径是否存在,若路径未设置会抛异常,且文件已存在则抛也异常。
3.通过配置文件获取到一个任务提交的临时目录,用来存放jar包、xml文件、以及切片信息,待所有必要信息都准备完了将 job 提交到yarn。
4.调用 getSplits()方法根据数据块的大小计算切片的大小,切片的个数决定了 MapTask 的个数,切片的时候会判断剩余文件的大小,若文件比切片大小的1.1倍大的时候才会进行切片,否则不进行切片,切的时候还是按照块大小切。
5.切片过后的数据通过 LineRecordReader 从 InputSplit中解析出KV值传给 Mapper。

说明：

1.AppMaster 通过切片的个数计算出相应的 MapTask 的个数,通过 ReduceTask 的个数计算出相应的分区个数。
2.切片时逻辑上的切片只是对文件做了个标记并没有对文件进行物理上的切分,之所以要求切片时剩余文件大小要大于片大小的1.1
倍是为了保证被切过后的文件都至少是片大小的0.1 倍,避免了小文件被切造成的资源浪费。
3. 1.1 只是用来判断是否对文件进行切片,而并非按1.1 倍去切,切片还是按照块大小进行切片, ‘bytesRemaining -= splitSize‘;
4.分块是物理上的切分,真实的将文件按128M进行了切分。
5.当某个文件只有一行数据时,且这一行数据特别大有100G,MapReduce并不会对其进行切片,默认使用TextInputformat不会对这一行数据切片。

CombinerTextInputFormat切片机制:

Hadoop默认使用TextInputFormat切片机制对任务进行切片,无论文件多大最终都会形成一个切片,这样就会产生大量的MapTask导致处理效率极低。
而CombinerTextInputFormat从逻辑上将多个小文件处理为一个切片,将这个切片交给一个MapTask处理。
<1>虚拟存储过程:将输入目录下所有的文件依次和设置的‘setMaxInputSplitSize‘进行比较,如果不大于设置的最大值,则逻辑上划分一块,如果大于设置值的的两倍可以按照最大值切一块,当剩余的值大于且不大于最大值的两倍时将剩余文件均分成两个虚拟存储块,这样可以防止存现很小的块。
<2>切片过程:按照存储过程对数据进行切块。

Mapper:

1.根据所需的业务逻辑对传进来的数据进行处理,然后通过 Context 将 KV值分组写出。
2.将数据传递入 OutputCollector。

Shuffle:

1.将 Mapper 传过来的数据存入环形缓冲区,右边存索引,左边存数据。当数据写入80%的时候开始反向溢写。
2.溢写时,先根据数据的分区排序,然后分区内再根据 Key 进行分组,最后用快排按 Key 值进行比较并对其相应的索排序,然后溢写出去,溢写的过程中可以使用 Conbiner对相同key的value进行合并,但不合并后不能改变原有的业务逻辑,待所有数据处理完后,通过轮询的方式对所有的溢写文件归并排序,默认每轮合并10个文件,该过程仍可以用 Conbiner。
3.最后将文件归并后的文件压缩后写入到磁盘。

Partition分区:

Hadoop默认使用Hash分区器,默认分区时根据Key的hashCode值再对ReduceTask的个数取模得到。

总结:

    <1>如果ReduceTask的数量大于分区数则会产生几个空的输出文件。
    <2>如果ReduceTask的数量(>1)大于分区数则有一部分数据会无处安放会报错。 
    <3>如果ReduceTask的数量=1,则不管MapTask输出多少个分区文件结果都只会交给这一个ReduceTask,并只会产生一个结果文件。
    <4>分区号必须从0开始,逐一累加。

Combiner和partition的作用:

Combiner:发生在map的最后一个阶段,对每个maptask的输出进行汇总,以减小网络传输量,缓解网络IO,增加Reducer的执行效率。
Partition:主要将map阶段产生的所有kv分配给不同的reducetask处理,可以讲reduce阶段处理的负载进行分摊。

Reducer:

1.ReduceTask根据自己的分区号，去各个MapTask机器上取相应的结果分区数据 ，通过网络IO的方式将文件从各个 MapTask 远程拷
贝到当前节点的内存中,ReduceTask 可以使用的资源上限为1G,如果ReduceTask实际使用的资源量超过该值，则会被强制杀死。从 MapTask
远程拷贝过来的文件会先加载到内存中缓存,该 Buffer 缓冲占内存的0.7 即700M用来缓存数据,另外300M用来计算。但由于溢写文件是
向磁盘中写入的速度远小于内存读取数据的速度，所以不会等到700M的缓存满了再溢写,而是当数据达到缓存的0.66 就开始溢。
当文件大小超出 0.66 倍缓冲内存,就会对部分数据进行溢出到磁盘并将剩余部分写入内存, 否则直接放到内存,最后将内存中的数据和磁盘
中的数据进行归并排序,在合并阶段ReduceTask启动了连个后台线程来进行文件合并,以防止内存或磁盘上的文件过多,最后根据Key值进行分组,
将相同的 Key分为一组。
2.将分组后的数据传递给 reduce() 方法,再通过 Context.write() 将文件进行写出。

说明：

1.ReduceTask的数量决定了分区的数量。两者相等时效率最高。
2.Combiner类似于MapTask和ReduceTask的中间件,不能替代Reduce,ReducerTask为0时,不走shuffle,就无combiner。

OutputFormat:

将ReduceTask的输出结果通过 LineRecordWriter , 把每条记录写为文本行,调用 toString() 方法转换成字符串进行输出。

说明：

1.切片的大小,默认采用块大小做为切片大小,当切片大小比块大小小时,可能会造成网络IO。

思考：如何用mapReduce实现一个TopN?

将map端输入每行数据按空格切分,转换成(用户,访问次数)的键值对,然后reduce端实现聚合,将结果写入用户、访问次数实体类,并实现排序,最后结果做TopN筛选。

MapReduce 优化：

 <1>小文件的处理:
          (1)Har归档。
          (2)CombineTextInputformat。
          (3)JVM重用。
       为什么要处理小文件？
	元数据层面：每个小文件都有一份元数据保存在Namenode内存中，无论文件多么小都会占150KB。小文件过多会占用 Namenode服务器大量内存,影响Namenode性能和使用寿命
	计算层面：默认情况下MR会对每个小文件启用一个Map任务计算，非常影响计算性能。同时也影响磁盘寻址时间
	128G 可以存储的文件块：128 * 1024*1024*1024 byte/150 byte = 9亿文件块
	关于JVM重用：无小文件不要开启JVM重用,因为会一直占用当前task的卡槽,直到任务完成后释放，开启JVM重用后可以在,mapred-site.xml 配置重用次数。
 <2> Map阶段:
	(1)增大环形缓冲区大小。由100m扩大到200m。
	(2)增大环形缓冲区溢写的比例。由80%扩大到90%。 
	(3)减少对溢写文件的merge次数。
	(4)不影响实际业务的前提下，采用Combiner提前合并，减少 I/O。
 <3>Reduce阶段:	
	(1)合理设置Map和Reduce个数:太少，会导致Task等待，延长处理时间；太多，会导致 Map、Reduce任务间竞争资源，造成处理超时等错误。
	(2)设置Map和Reduce共存：调整 slowstart.completedmaps 参数,约为5% 使Map运行到一定程度后，Reduce也开始运行，减少Reduce的等待时间。
		注意：设置Map和Reduce共存,有条件不可以影响最终业务逻辑,如求平均值得等所有map完成后才能Reduce。
	(3)规避使用Reduce，因为Reduce在用于连接数据集的时候将会产生大量的网络消耗。
	(4)增加每个Reduce去Map中拿数据的并行数。
	(5)集群性能有保障的前提下，增大Reduce端存储数据内存的大小。
	(6)合理设置Reduce端的Buffer,Buffer为Reduce内存大小的 0.7,当达到Buffer的阈值 0.66 时会溢写到磁盘,然后Reduce会将磁盘中与内存中
	   的数据进行归并,该过程可能存在多个多次写磁盘->读磁盘的过程,会有大量的网络IO,可以通过设置使时,Buffer中的部分数据直接输送到Reducer,
	   从而减小网络IO。
 <4>IO传输:
	采用数据压缩的方式，减少网络IO的的时间。
	(1)在map输入端主要考虑数据量大小和切片,支持切片的有Bzip2、LZO。注意：LZO要想支持切片必须创建索引；
	(2)而map输出端主要考虑速度，速度快的snappy、LZO；
	(3)在reduce输出端主要看具体需求，例如作为下一个mr输入需要考虑切片，永久保存考虑压缩率比较大的gzip。
 <5>整体:
	(1)增加Namenode的内存
	(2)增大单个任务默认内存
	(3)增加MapTask和ReduceTask的可用内存上限
	(4)增加MapTask和ReduceTask的堆内存大小
	(5)增加MapTask和ReduceTask的CPU核数
	(6)增加每个Container的CPU核数和内存大小
	(7)配置DataNode的多目录存储
	(8)增大NameNode线程池处理DataNode并发心跳以及客户端并发元数据操作的个数。

 <6>数据倾斜：
	数据倾斜现象：
		1.数据频率倾斜--某个分区的数据量远大于其他分区的数据量。
		2.数据大小倾斜--部分记录的大小远大于平均值
	(1)抽样和范围分区：通过对原始数据进行抽样后得到的结果在进行预分区。
	(2)自定义分区器：根据输出键进行自定义分区。
	(3)使用Combiner：使用Combiner可以大量减小数据倾斜,可能的情况下,Combiner的目的就是聚合精简数据。
	(4)使用MapJoin,避免ReduceJoin：MapJoin将小表以集合的形式预加载到内存中,然后再与大表Join,并且没有Reduce阶段。
 <7>Hadoop宕机
	(1)MR造成系统宕机:需要控制yarn同时运行的任务数,以及每个任务申请的最大内存。
	(2)文件写入过快造成NameNode宕机:增大Kafka的存储大小,减小HDFS的写入速度。

Hadoop 之 MapReduce

原文：https://www.cnblogs.com/yuexiuping/p/14810566.html