mapreduce 的过程

el/2024/7/13 10:58:41

    从前面的例子我们已经大致了解了一个MapReduce的作业的过程,但是这样是不够的,本节将深入探讨MapReduce的整个过程。

  • 从输入到输出

    从前而的WordCount可以看出,一个MapReduce作业经过了input, map, combine, reduce.output五个阶段,其中combine阶段并不一定发生,map输出中间结果被分发到 reducer 的过程被称为shuffle(数据混洗),如下图所示。

    在shuffle阶段还会发生copy (复制)和Sort(排序)
    在MapReduce的过程中,一个作业被分成Map和Reduce计算两个阶段,它们分别由一个或者多个Map任务和Reduce任务组成。如下图示,一个MapReduce作业从数据的流向可以被划分为Map任务和 Reduce任务。正如前而所说,当用户向 Hadoop提交一个MapReduce作业时,JobTracker则会根据各个TaskTracker周期性发送过来的心跳信息、综合考虑TaskTracker的资源剩余量、作业优先级、作业提交时间等因素,为TaskTracke分配合适的任务Reduce任务默认会在Map任务数量完成5%后才开始启动。



    Map任务的执行过程可概括为:首先通过用户指定的InputFormat类(如WordCount中的FilelnputFormat类)中的getSplits方法和next方法将输入文件切片并解析成键值对作为map函数的输入。然后map函数经过处理之后输出并将中间结果交给指定的Partitioner处理,确保中间结果分发到指定的Reduce任务处理,此时如果用户指定了Combiner,将执行combine操作,最后map函数将中间结果保存到本地。
    Reduce任务的执行过程可概括为:首先击要将已经完成的Map任务的中间结果复制到Reduce任务所在的节点,待数据复制完成后,再以key进行排序,通过排序,将所有key相同的数据交给reduce函数处理,处理完成后,结果直接输出到HDFS  上。


  • input

    如果使用HDFS上的文件作为MapReduce的输入(由于用户的数据大部分数据是以文件的形式形式存储在HDFS上,所以这是最常见的情况) MapReduce计算框架首先会用
org.apache.hadoop.mapreduce.InputFormat类的子类FileInputFormat类将作为输入的HDFS上的文件切分形成输入分片(InputSplit ),每个InputSplit将作为一个Map任务的输入,再将InputSplit解析为键值对。InputSplit的大小和数量对于MapReduce作业的性能有非常大的影响,因此有必要深入了解InputSplit。

    InputSplit只是逻辑上对输入数据进行分片,并不会将文件在磁盘上切成分片进行存储。InputSplit只一记录了分片的元数据节点信息,例如起始位置、长度以及所在的节点列表等。数据切分的算法需要确定InputSplit的个数,对于HDFS上的文件,FileInputFormat类使用computeSplitSize方法计算出InputSplit的大小,代码如下:



protected long computeSplitSize(long blockSize long minSize, long maxSize)
{return Math.max(minSize,Math.min(maxSize, blockSize))
}

     其中minSize由mapred-site.xml文件中的配置项mapred.min.split.size决定,默认为1;

           maxSize由mapred-site.xml文件中的配置项mapred.max.split.size决定,默认为9 223 372 036 854 775 807;

          而blockSize也是由hdfs-site.xml文件中的配置项dfs.block.size决定,默认为67 108 864字节(64 MB )。

  所以InputSplit的大小的确定公式为:

        max(mapred.min.split.size ,Math.min(mapred.max.split.size,dfs .block.size))

       代码理解:  从mapred.max.split.size与dfs .block.size当中选最小的来和mapred.min.split.size一起确定InputSplit的大小

一般来说,dfs.block.size的大小是确定不变的,所以得到目标InputSplit大小,只需改变mapred.min.split.size和InputSpIitSizeo InputSplitmapred.max.split.size的大小即可。InputSplit的数量为文件大小除以的元数据信息会通过以下代码取得:

splits .add (new FileSplit(path, lengthbytesRemaining, splitSize,blkLocations[blkIndex].getHosts()));
    从上面的代码可以发现,元数据的信息由4部分组成:文件路径、文件开始的位置、文件结束的位置、数据块所在的host。

    对于Map任务来说,处理的单位为一个InputSplitt。而InputSplit是一个逻辑概念,InputSplit所包含的数据是仍然是存储在HDFS的块里面,它们之间的关系如下图所示。

    InputSplit可以不和块对齐,根据前面的公式也可以看出,一个InputSplit的大小可以大于一个块的大小亦可以小于一个块的大小。Hadoop在进行任务调度的时候,会优先考虑本节点的数据,如果本节点没有可处理的数据或者是还需要其他节点的数据,Map任务所在的节点会从其他节点将数据通过网络传输给自己。当InputSplit的容量大于块的容量,Map任务就必须从其他节点读取一部分数据,这样就不能实现完全数据本地性,所以当使用FilelnputFormat实现lnputFormat时,应尽量使InputSplit的大小和块的大小相同以提高Map任务计算的数据本地性。

    当输入文件切分为InputSplit后,由FilelnputFormat 的类(如 TextlnputFormat 的createRecordReader方法将InputSplit解析为键值对,代码如下:

public RecordReader<LongWritable, Text>
createRecordReader(InputSplit split,TaskAttemptContext context) {
String delimiter = context.getConfiguration ().get ("textinputformac .record.delimiter"); byte[] recordDelimiterBytes » null;if (null !- delimiter)
recordDelimiterBytes - delimiter.getBytes();return new LineRecordReader(recordDelimiterBytes);
}

     此处默认是将行号作为键。解析出来的键值对将被用来作为map函数的输入。至此input阶段结束。


  • map及中间结果的输出

    InputSplit将解析好的键值对交给用户编写的map函数处理,处理的后的中间}结果会写到本地磁盘上,在刷写磁盘的过程中,还做了partition(分区)和sort(排序)的操作
    map函数产生输出时,并不是简单地刷写磁盘。为了保证I/O效率,采取了先写到内存的环形缓冲区,并作一次预排序,如下图所示。


    每个Map任务都有一个环形内存缓冲区,用于存储map函数的输出。默认情况下,缓冲区的大小是100 MB,该值可以通过mapred-site.xml文件的io.sort.mb的配置项配置。一旦缓冲区内容达到阈值(由mapred-site.xml文件的io.sort.spill.percent的值决定,默认为0.80或80% ),一个后台线程便会将缓冲区的内容溢写( spill)到磁盘中。在写磁盘的过程中,map函数的输出继续被写到缓冲区,但如果在此期间缓冲区被填满,map会阻塞直到写磁盘过程完成。写磁盘会以轮询的方式写到mapred.local.dir ( mapred-site.xml文件的配置项)配置的作业特定目录下。

    在写磁盘之前,线程会根据数据最终要传送到的Reducer把缓冲区的数据划分成(默认是按照键)相应的分区。在每个分中,后台线程按键进行内排序,此时如果有一个Combiner.它会在排序后的输出上运行。

    一旦内存缓冲区达到溢出的阀值,就会新建一个溢出写文件,因此在Map任务写完其最后 一个输出记录之后,会有若干个溢出写文件。在Map任务完成之前,溢出写文件被合并成一个已分区且已排序的输出文件作为map输出的中问结果,这也是Map任务的输出结果。
    如果己经指定Combiner且溢出写次数至少为3时,Combiner就会在输出文件写到磁盘之前运行。如前文所述,Combiner可以多次运行,并不影响输出结果。运行Combiner的意义在于使map输出的中问结果更紧凑,使得写到本地磁盘和传给Reducer的数据更少。
    为了提高磁盘I/O性能,可以考虑压缩map的输出,这样会让写磁盘的速度更快,节约磁盘空间,从而使传送给Reducer的数据量减少。默认情况下,map的输出是不压缩的,但只要将mapred-site.xml文件的配置项mapred.compress.map.output设为true即可开启压缩功能。使用的压缩库由mapred-site.xml文件的配置mapred.map.output.compression.codec指定。下表列出了目前Hadoop支持的压缩格式。





    map输出的中间结果存储的格式为IFile, IFile足一种支持行压缩的存储格式,支持上述压缩算法。
    Reducer通过HTTP方式得到输出文件分区。将map输出的中间结果(map输出)发送到Reducer的工作线程的数量由mapred-site.xml文件tasktracker.http.threads配置项决定,此配置针对每个节点,即协个TaskTracker,  而不是每个 Map任务,默认是40, 可以根据作业大小,集群规模以及节点的计算能力而增大。

  •    shuffle

    shuffle,也叫数据混洗。在某此语境}h,代表map函数产生输出到reduce的消化输入的整个过程,在本节中,我们将其理解为只代表Reduce任务获取Map任务输出 ( map output)的这部分过程。 shuffle是MapReduce 非常重要的一部分,理解其原理有助于优化MapReduce程序。
    如上一节所述,Map任务输出的结果位于运行Map任务的TaskTracker所在的节点的本地磁盘上。TaskTracker需要为这些分区文件(map输出)运行Reduce任务。但是,Reduce任务可能需要多个Map任务的输出作为其特殊的分区文件。每个Map任务的完成时间可能不同.当只要有一个任务完成,Reduce任务就开始复制其输出。这就是shuffle中的copy阶段,如图所示。Reduce任务有少量复制线程,可以井行取得Map任务的输出,默认值是5个线程,该值可以通过设置mapred-site.xml的mapred.reduce.parallel.copies配置项来改变。
    如果map输出相当小,则会被复制到 Reducer所在的TaskTracker的内存的缓冲区中,缓冲区的大小由mapred-site.xml文件中的mapred.job.shuffle.input.buffer.percent 配置项指定。否则,,map输出将会被复制到磁盘。一旦内存缓冲区达到阀值大小(由mapred-site.xml文件的mapred.job.shuffle.merge.percent配置项决定)或缓冲区的文件数达到阈值人小(由mapred-site.xml文件的mapred.inmem.merge.threshold配置项决定),则合井后溢写到磁盘中。
    随着溢写到磁盘的文件增多,后台线程会将它们合并为更大的、有序的文件,这会为后面的合并节省时间。为了合并,压缩的中间结果都将在内存中解压缩。
    复制完所有的map输出,shuffle进入sort阶段。这个阶段将合并map的输出文件.并维持其顺序排序,其实做的是归并排序。排序的过程是循环进行,如果有50个map的输出文件,而合并因子(由mapred-site.xml文件的io.sort.factor配置项决定,默认为10)为 10,合并操们将进行5次,每次将10个文件合并成一个文件,最后会有5个文件,这5个文件由于.不满足合并条件(文件数小于合并因子).则不会进行合并,将会直接把这5个文件交给reduce函数处理。至此,shuffle阶段完成。


    从shuffle的过程可以看出,Map任务处理的是一个InputSplit,而Reduce任务处理的是所有Map任务同一个分区的中间结果。

  • reduce及最后结果的输出

    reduce阶段操作的实质就是对经过shuffle处理后的文件调用reduce函数处理。由于经过了shuffle的处理,文件都是按键分区且有序,对相同分区的文件调用一次reduce函数处理。
    与map的中间结果不同的是,reduce的输出一般为HDFS

  • sort

       排序贯穿于Map任务和Reduce任务,是MapReduce非常重要的一环,排序操作属于MapReduce计算框架的默认行为,不管流程是否需要,都会进行排序。在MapReduce计算框架中,主要用到了两种排序算法:快速排序和归并排序

      (1)快速排序:通过一趟排序将要排序的数据分割成独立的两部分,其中一部分的所有数据比另外一部分的所有数据都要小,然后再按此方法对这两部分数据分别进行快速排序,整个排序过程可以递归进行,以此达到整个数据变成有序序列。
      (2)归并排序:归并排序在分布式计算里面用得非常多,归并排序本身就是一个采用分治法的典型应用。归并排序是将两个(或两个以上)有序表合并成一个新的有序表,即把待排序序列分为若干个有序的护序列,再把有序的子序列合并为整体有序序列
  

     在Map任务和Reduce任务的过程中,一共发生了3次排序操作
     (1)当map函数产生输出时,会首先写入内存的环形缓冲区,当达到设定的阀值,在刷写磁盘之前,后台线程会将缓冲区的数据划分成相应的分区。在每个分区中,后台线程按键进行内排序,如下图所示。



    (2)在Map任务完成之前,磁盘上存在多个己经分好区,并排好序的、大小和缓冲区一样的溢写文件,这时溢写文件将被合并成一个己分区且已排序的输出文件。由于溢写文件已经经过第一次排序,所以合并文件时只需再做一次排序就可使输出文件整体有序。
    (3)在shuffle阶段,需要将多个Map任务的输出文件合井,由于经过第二次排序,所以合并文件时只需再做一次排序就可使输出文件整体有序,如下图B所示





在这3次排序中第一次是在内存缓冲区做的内排序,使用的算法是快速排序,第二次排序和第三次排序都是在文件合并阶段发生的,使用的是归并排序。


  • 作业的进度组成

       一个MapReduce作业在Hadoop 上运行时,客户端的屏幕通常会打印作业日志,如下:

        对一个大型的MapReduce作业来说(比如说数据清洗),执行时间可能会比较长,通过日志了解作业的运行状态和作业进度是非常重要的。
       对于map来说,进度代表实际处理输入所占比例,例如map 60% reduce 0%表示Map任务己经处理了作业输入文件的60%,而Reduce任务还没有开始。而对于reduce的进度来说,情况就比较复杂,从前面得知,reduce阶段分为copy, sort和reduce,这3个步骤共同组成了reduce的进度,各占1/3。如果Reduce己经处理了2/3的输入,那么整个Reduce的进度应该是1/3十1/3+ 1 /3 X (2/3) = 5/9,因为reduce开始处理输入时,copy和sort已经完成。







http://www.ngui.cc/el/5432318.html

相关文章

linux当中使用vi/vim编辑器时,错误操作遇到警告信息,该如何做。

最近尼玛哥在使用linux 的时候&#xff0c; 尤其是编译文件时进入编辑模式后&#xff0c;编写了一堆文件&#xff0c;突然又不想编写了&#xff0c;或者是因为文件太多&#xff0c;而又不希望在一个一个倒退&#xff0c;直接按下了ctrl z 键&#xff0c;退出文件编辑。这时候&…

MySQL 5.5 Command Line Client 打开出现闪退(错误)的解决办法

解决方法&#xff0c;打开文件安装所在位置。 假设尼玛哥的文件存储在 E:\mysqlserver\mysql 目录下&#xff0c;在此目录当中&#xff0c;咱们发现&#xff0c;有6个ini结尾的文件。 此时&#xff0c;咱们需要做的就是&#xff0c;复制任意一个 ini结尾的文件&#xff…

修改mysql默认编码重启后又还原,且在修改my.ini配置文件出现1067错误的解决办法。

尼玛哥在命令行操作来修改mysql的编码的时候&#xff0c;每次重新启动&#xff0c;都会还原到原有的配置去&#xff0c;无数次的失败&#xff0c;当出现这种情况时这时候&#xff0c;你要做的就是去修改配置文件了 注意&#xff0c;配置文件放在你所安装的mysql里边&#xff0…

Neo4j CQL - (6)-关系基础

Neo4j图数据库遵循属性图模型来存储和管理其数据。 根据属性图模型&#xff0c;关系应该是定向的。 否则&#xff0c;Neo4j将抛出一个错误消息。 基于方向性&#xff0c;Neo4j关系被分为两种主要类型。 单向关系双向关系 在以下场景中&#xff0c;我们可以使用Neo4j CQL C…

Python ( 1 ) ----- 简介

Python是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本语言。 Python的设计具有很强的可读性&#xff0c;相比其他语言经常使用英文关键字&#xff0c;其他语言的一些标点符号&#xff0c;它具有比其他语言更有特色语法结构。 Python 是一种解释型语言&#xff…

正则表达式 分支条件

又叫逻辑运算符&#xff0c;在此X和Y表示两个表达式 XY X紧跟YX|Y 表示X或Y&#xff0c;从左到右&#xff0c;满足第一个条件就不会继续匹配了。 文章参考&#xff1a;https://www.w3cschool.cn/rxoyts/l17fcozt.html

windows 环境下在anaconda 3中安装python2和python3两个环境(python2和python3共存)

博主本人是在学习机器学习的过程当中&#xff0c;看到的相关书籍使用的python 版本不一&#xff0c;但又想将每个版本都懂能使用一番&#xff0c;接下来&#xff0c;博主给大家介绍一个比较简单方便的方式&#xff1a; 首先&#xff1a;安装python 环境&#xff0c;极力推荐使…

pycharm当中,如何导入python2或者python3环境

step 1 step 2 step 3 OK ,环境添加完毕&#xff0c;此环境为python 2.7.17&#xff08;python 3可按此步骤添加&#xff09;&#xff0c;如此一来&#xff0c;我们可以灵活的再此处切换所有的编译环境。

数据挖掘面向什么类型的应用?

作为一个应用驱动的学科&#xff0c;数据挖掘已经在许多应用中获得巨大成功。我们不可能一一枚举数据挖掘扮演关键角色的所有应用。在知识密集的应用领域&#xff0c;如生物信息学和软件工程&#xff0c;数据挖掘的表现更需要深人处理&#xff0c;这已经超出本书的范围。应用作…

用pycharm出现,ImportError: No module named model_selection

首先&#xff0c;先说明&#xff0c;本人出现这个问题&#xff0c;是因为需要使用 train_test_split 方法&#xff0c;具体的引用是&#xff1a; from sklearn.model_selection import train_test_split 出现了这个错误&#xff1a; ImportError: No module named model_sel…