MapReduce计算的流程-Hadoop

输入 --> map --> shuffle --> reduce -->输出

MapReduce计算的流程-Hadoop

1 map端

  •        分块

        针对上传到hdfs上的文件,会针对文件进行分块(物理分块,固定大小64M,多余的则进行分,少的则不变),分块之后存储相应的datanode计算机的磁盘上面;

         把File划分成Block,这个是物理上真真实实的进行了划分,数据文件上传到HDFS里的时候,需要划分成一块一块,每块的大小由hadoop-default.xml里配置选项进行划分。一个大文件可以把划分后的所有块存储到同一个磁盘上,也可以在每个磁盘上都存在这个文件的分块。

  •          分片

         片,相当于一个文件逻辑的量,目的只是为了让map task更好地获取数据。hadoop中的InputFormat接口中的getSplit()方法得到的。如果片的大小设置与块的大小相同的话,那么有多少块就有多少的片;片的多少与map的数量相对应;

        一个片为一个splits,即一个map,只要搞清楚片的大小,就能计算出运行时的map数。而一个split的大小是由goalSize, minSize, blockSize这三个值决定的。computeSplitSize的逻辑是,先从goalSize和blockSize两个值中选出最小的那个(比如一般不设置map数,这时blockSize为当前文件的块size,而goalSize是文件大小除以用户设置的map数得到的,如果没设置的话,默认是1),在默认的大多数情况下,blockSize比较小。然后再取blockSize和minSize中最大的那个。而minSize如果不通过”mapred.min.split.size”设置的话(”mapred.min.split.size”默认为0),minSize为1,这样得出的一个splits的size就是blockSize,即一个块一个map,有多少块就有多少map。

       在FileInputFormat中,计算切片大小的逻辑:Math.max(minSize, Math.min(maxSize, blockSize))

  • patition
  • Partition主要作用就是将map的结果发送到相应的reduce。这就对partition有两个要求:

    1)均衡负载,尽量的将工作均匀的分配给不同的reduce。

    2)效率,分配速度一定要快。

partition算法是可以人为定义的,默认的是HashPartitioner,用key的哈希值对reduce个数求模(key.hascode%numofreduce),同一个key的值一般都会在同一个partition的;

  • 环形存储区

经过map函数的逻辑处理后的数据输出之后,会通过OutPutCollector收集器将数据收集到环形缓存区保存。

环形缓存区的大小默认为100M,当保存的数据达到80%时,就将缓存区的数据溢出到磁盘上保存。

map阶段的输出结果会先写到内存缓冲区,然后由缓冲区写到磁盘上。默认的缓冲区大小是100M,溢出的百分比是0.8,也就是说当缓冲区中达到80M的时候就会往磁盘上写。如果map计算完成后的中间结果没有达到80M,最终也是要写到磁盘上的,因为它最终还是要形成文件。那么,在往磁盘上写的时候会进行分区和排序(可能会按照值的大小进行排序)。一个map的输出可能有多个这个的文件,这些文件最终会合并成一个,这就是这个map的输出文件。MapReduce计算的流程-Hadoop

这个数据结构其实就是个字节数组byte[],叫Kvbuffer,名如其义,但是这里面不光放置了数据,还放置了一些索引数据,给放置索引数据的区域起了一个Kvmeta的别名,在Kvbuffer的一块区域上穿了一个IntBuffer(字节序采用的是平台自身的字节序)
            的马甲。数据区域和索引数据区域在Kvbuffer中是相邻不重叠的两个区域,用一个分界点来划分两者,分界点不是亘古不变的,而是每次Spill之后都会更新一次。初始的分界点是0,数据的存储方向是向上增长,索引数据的存储方向是向下增长.
            
            注意:上述的分界点就可以理解图中的赤道信息
            
            Kvbuffer的存放指针bufindex(即数据的存储方向)是一直闷着头地向上增长,比如bufindex初始值为0,一个Int型的key写完之后,bufindex增长为4,一个Int型的value写完之后,bufindex增长为8。 (int型的数据占有4个字节)
            
            索引是对在kvbuffer中的键值对的索引,是个四元组,包括:value的起始位置、key的起始位置、partition值、value的长度,占用四个Int长度,Kvmeta的存放指针Kvindex每次都是向下跳四个“格子”,然后再向上一个格子一个格子地填充四元组
            的数据。比如Kvindex初始位置是-4,当第一个键值对写完之后,(Kvindex+0)的位置存放value的起始位置、(Kvindex+1)的位置存放key的起始位置、(Kvindex+2)的位置存放partition的值、(Kvindex+3)的位置存放value的长度,然后Kvindex跳到-8位置,等第二个键值对和索引写完之后,Kvindex跳到-12位置。

  • 合并Combiner

如果指定了Combiner,可能在两个地方被调用: 
1.当为作业设置Combiner类后,缓存溢出线程将缓存存放到磁盘时,就会调用; 
2.缓存溢出的数量超过mapreduce.map.combine.minspills(默认3)时,在缓存溢出文件合并的时候会调用

合并(Combine)和归并(Merge)的区别: 
两个键值对<“a”,1>和<“a”,1>,如果合并,会得到<“a”,2>,如果归并,会得到<“a”,<1,1>>

特殊情况:当数据量很小,达不到缓冲区阙值时,怎么处理?

对于这种情况,目前看到有两种不一样的说法:

       ①不会有写临时文件到磁盘的操作,也不会有后面的合并。

       ②最终也会以临时文件的形式存储到本地磁盘

  • 归并merge

当一个map task处理的数据很大,以至于超过缓冲区内存时,就会生成多个spill文件。此时就需要对同一个map任务产生的多个spill文件进行归并生成最终的一个已分区且已排序的大文件。配置属性mapreduce.task.io.sort.factor控制着一次最多能合并多少流,默认值是10。这个过程包括排序和合并(可选),归并得到的文件内键值对有可能拥有相同的key,这个过程如果client设置过Combiner,也会合并相同的key值的键值对(根据上面提到的combine的调用时机可知)。

溢出写文件归并完毕后,Map将删除所有的临时溢出写文件,并告知NodeManager任务已完成,只要其中一个MapTask完成,ReduceTask就开始复制它的输出(Copy阶段分区输出文件通过http的方式提供给reducer)

  • 压缩 

写磁盘时压缩map端的输出,因为这样会让写磁盘的速度更快,节约磁盘空间,并减少传给reducer的数据量。默认情况下,输出是不压缩的(将mapreduce.map.output.compress设置为true即可启动)

 2 Reduce端

①复制copy

②归并merge

③reduce

MapReduce计算的流程-Hadoop

①复制copy

Reduce进程启动一些数据copy线程,通过HTTP方式请求MapTask所在的NodeManager以获取输出文件。 
NodeManager需要为分区文件运行reduce任务。并且reduce任务需要集群上若干个map任务的map输出作为其特殊的分区文件。而每个map任务的完成时间可能不同,因此只要有一个任务完成,reduce任务就开始复制其输出。

reduce任务有少量复制线程,因此能够并行取得map输出。默认线程数为5,但这个默认值可以通过mapreduce.reduce.shuffle.parallelcopies属性进行设置。

【Reducer如何知道自己应该处理哪些数据呢?】 
因为Map端进行partition的时候,实际上就相当于指定了每个Reducer要处理的数据(partition就对应了Reducer),所以Reducer在拷贝数据的时候只需拷贝与自己对应的partition中的数据即可。每个Reducer会处理一个或者多个partition。

【reducer如何知道要从哪台机器上去的map输出呢?】 
map任务完成后,它们会使用心跳机制通知它们的application master、因此对于指定作业,application master知道map输出和主机位置之间的映射关系。reducer中的一个线程定期询问master以便获取map输出主机的位置。知道获得所有输出位置。

②归并merge

 Copy 过来的数据会先放入内存缓冲区中,这里的缓冲区大小要比 map 端的更为灵活,它基于 JVM 的 heap size 设置,因为 Shuffle 阶段 Reducer 不运行,所以应该把绝大部分的内存都给 Shuffle 用。

Copy过来的数据会先放入内存缓冲区中,如果内存缓冲区中能放得下这次数据的话就直接把数据写到内存中,即内存到内存merge。Reduce要向每个Map去拖取数据,在内存中每个Map对应一块数据,当内存缓存区中存储的Map数据占用空间达到一定程度的时候,开始启动内存中merge,把内存中的数据merge输出到磁盘上一个文件中,即内存到磁盘merge。与map端的溢写类似,在将buffer中多个map输出合并写入磁盘之前,如果设置了Combiner,则会化简压缩合并的map输出。Reduce的内存缓冲区可通过mapred.job.shuffle.input.buffer.percent配置,默认是JVM的heap size的70%。内存到磁盘merge的启动门限可以通过mapred.job.shuffle.merge.percent配置,默认是66%。

当属于该reducer的map输出全部拷贝完成,则会在reducer上生成多个文件(如果拖取的所有map数据总量都没有内存缓冲区,则数据就只存在于内存中),这时开始执行合并操作,即磁盘到磁盘merge,Map的输出数据已经是有序的,Merge进行一次合并排序,所谓Reduce端的sort过程就是这个合并的过程,采取的排序方法跟map阶段不同,因为每个map端传过来的数据是排好序的,因此众多排好序的map输出文件在reduce端进行合并时采用的是归并排序,针对键进行归并排序。一般Reduce是一边copy一边sort,即copy和sort两个阶段是重叠而不是完全分开的。最终Reduce shuffle过程会输出一个整体有序的数据块。

③reduce

当一个reduce任务完成全部的复制和排序后,就会针对已根据键排好序的Key构造对应的Value迭代器。这时就要用到分组,默认的根据键分组,自定义的可是使用 job.setGroupingComparatorClass()方法设置分组函数类。对于默认分组来说,只要这个比较器比较的两个Key相同,它们就属于同一组,它们的 Value就会放在一个Value迭代器,而这个迭代器的Key使用属于同一个组的所有Key的第一个Key。

在reduce阶段,reduce()方法的输入是所有的Key和它的Value迭代器。此阶段的输出直接写到输出文件系统,一般为HDFS。如果采用HDFS,由于NodeManager也运行数据节点,所以第一个块副本将被写到本地磁盘。

1、当reduce将所有的map上对应自己partition的数据下载完成后,reducetask真正进入reduce函数的计算阶段。由于reduce计算时同样是需要内存作为buffer,可以用mapreduce.reduce.input.buffer.percent(default 0.0)(源代码MergeManagerImpl.java:674行)来设置reduce的缓存。

这个参数默认情况下为0,也就是说,reduce是全部从磁盘开始读处理数据。如果这个参数大于0,那么就会有一定量的数据被缓存在内存并输送给reduce,当reduce计算逻辑消耗内存很小时,可以分一部分内存用来缓存数据,可以提升计算的速度。所以默认情况下都是从磁盘读取数据,如果内存足够大的话,务必设置该参数让reduce直接从缓存读数据,这样做就有点Spark Cache的感觉。

分组和上面提到的partition(分区)不同,分组发生在reduce端,reduce的输入数据,会根据key是否相等而分为一组,如果key相等的,则这些key所对应的value值会作为一个迭代器对象传给reduce函数。以单词统计为例,reduce输入的数据就如:第一组:(a,(1,3,5,3,1))第二组:(b,(6,2,3,1,5))。上述例子也可以看出在map端是执行过combiner函数的,否则reduce获得的输入数据是:第一组:(a,(1,1,1,1,1,...))第二组:(b,(1,1,1,1,1...))。对每一组数据调用一次reduce函数。

2、Reduce在这个阶段,框架为已分组的输入数据中的每个键值对对调用一次 reduce(WritableComparable,Iterator, OutputCollector, Reporter)方法。Reduce任务的输出通常是通过调用 OutputCollector.collect(WritableComparable,Writable)写入文件系统的。

关于分组的深入理解,请看这篇文章:https://mp.csdn.net/postedit/81778972

理解还不是很深刻,需要实践来检验和探索。如有错误,请大家不吝指正!!!

参考文章:

https://blog.csdn.net/u014374284/article/details/49205885 (相对来说,感觉这篇写的最好)

https://blog.csdn.net/lb812913059/article/details/79899644

https://blog.csdn.net/lb812913059/article/details/79899798

https://www.cnblogs.com/DianaCody/p/5425658.html
https://blog.csdn.net/asn_forever/article/details/81233547

and so....