“深入简出”好程序员教你MapReduce全过程

Posted 2021-11-30 gcghcxy

　　欲说MapReduce，好程序员要和大家先讲一对概念——移动数据和移动计算。

　　在学习大数据的时候，我们便会接触移动数据和移动计算，这两种联系紧密而又有很大不同的概念，其中移动计算也叫做本地计算。

　　在以前的数据处理中时使用的移动数据，其实就是将需要处理的数据传输到存放不同处理数据方式逻辑的各个节点上。这样做的效率很低，特别是大数据中的数据量是很大的，至少都是GB以上，更大的是TB、PB甚至更大，而且磁盘I/O、网络I/O的效率是很低的，这样处理起来就需要很长的时间，远远不能满足我们的要求。而移动计算就出现了。

　　移动计算，也叫做本地计算，是数据就存放在节点上不再变动，而是将处理逻辑程序传输到各个数据节点上。由于处理程序的大小肯定不会特别的大，这样就可以实现很快将程序传输到存放数据的各个节点上去，然后本地执行处理数据，效率高。现在的大数据处理技术都是采用这种方式。

　　那么讲MapReduce的过程，言简意赅的说，我们分两个阶段：

　　1.Map阶段：

　　（1）Read：读取数据源，将数据进行filter成一个个的K/V。

　　（2）Map：在map函数中，处理解析的K/V，并产生新的K/V。

　　（3）Collect：输出结果，存于环形内缓冲区。

　　（4）Spill：内存区满，数据写到本地磁盘，并生产临时文件。

　　（5）Combine：合并临时文件，确保生产一个数据文件。

　　2.Reduce阶段：

　　（1）Shuffle：Copy阶段，Reduce Task到各个Map Task远程复制一分数据，针对某一份数据，若其大小超过一定阀值，则写磁盘；否则放到内存。

　　（2）Merge：合并内存和磁盘上的文件，防止内存占用过多或磁盘文件过多。

　　（3）Sort：Map Task阶段进行局部排序，Reduce Task阶段进行一次归并排序。

　　（4）Reduce：将数据给reduce函数。

　　（5）Write：reduce函数将其计算的结果写到HDFS上。

　　对其进行深度的解析，我们可以分几个阶段进行解释：

　　一、MapTask阶段

　　（1）Read阶段：MapTask通过用户编写的RecordReader，从输入InputSplit中解析出一个个key/value。

　　（2）Map阶段：该节点主要是将解析出的key/value交给用户编写map()函数处理，并产生一系列新的key/value。

　　（3）Collect收集阶段：在用户编写map()函数中，当数据处理完成后，一般会调用 OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用 Partitioner），并写入一个环形内存缓冲区中。

　　（4）Spill阶段：即“溢写”，当环形缓冲区满后，MapReduce 会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。

　　这里要注意溢写的情况，我们不妨按照以下步骤进行处理：

　　步骤1：利用快速排序算法对缓存区内的数据进行排序，排序方式是，先按照分区编号partition进行排序，然后按照key进行排序。这样，经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照key有序。

　　步骤2：按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.out（N表示当前溢写次数）中。如果用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次聚集操作。

　　步骤3：将分区数据的元信息写到内存索引数据结构SpillRecord中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件output/spillN.out.index中。

　　当遇到Combine问题时，在所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件。当所有数据处理完后，MapTask会将所有临时文件合并成一个大文件，并保存到文件output/file.out中，同时生成相应的索引文件output/file.out.index。在进行文件合并过程中，MapTask以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并io.sort.factor（默认100）个文件，并将产生的文件重新加入待合并列表中，对文件排序后，重复以上过程，直到最终得到一个大文件。让每个MapTask最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件output/spillN.out.index中。

　　二、Shuffle阶段(map端的输出到reduce的输入)

　　（1）maptask收集我们的map()方法输出的kv对，放到内存缓冲区中。

　　（2）从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件 。

　　（3）多个溢出文件会被合并成大的溢出文件 。

　　（4）在溢出过程中，及合并的过程中，都要调用partitioner进行分区和针对key进行排序 。

　　（5）reducetask根据自己的分区号，去各个maptask机器上取相应的结果分区数据。

　　（6）reducetask会取到同一个分区的来自不同maptask的结果文件，reducetask会将这些文件再进行合并（归并排序） 。

　　（7）合并成大文件后，shuffle的过程也就结束了，后面进入reducetask的逻辑运算过程。（从文件中取出一个一个的键值对group，调用用户自定义的reduce()方法）

　　注意Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快。缓冲区的大小可以通过参数调整，参数：io.sort.mb默认100M。

　　三、ReduceTask阶段

　　（1）Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。

　　（2）Merge阶段：在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。

　　（3）Sort阶段：按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。

　　（4）Reduce阶段：reduce()函数将计算结果写到HDFS上。

　　简单精炼，深入精析，你要的全都给你。好程序员教你学好MapReduce，搞定大数据！更多大数据相关知识，稍后呈现……