MapReduce运行流程分析

研究MapReduce已经有一段时间了。起初是从分析WordCount程序开始,后来开始阅读Hadoop源码,自认为已经看清MapReduce的运行流程。现在把自己的理解贴出来,与大家分享,欢迎纠错。

还是以最经典的WordCount程序作为基础,来分析map阶段、reduce阶段和最复杂的shuffle阶段。

文本1:hello world                                      文本2:map reduce

hello hadoop                                              java interface

abc qaz                                                      java hdfs

     java jvm                                                    spark storm

这样的2个小文本文件(不足64M),肯定会产生2个map任务,reduce任务默认是1个。当然,map任务和reduce任务的个数都可以在程序中或者配置文件中人为设置。为了说明partition的过程,我们把reduce任务的个数设为2。

1、map阶段

map1                                                            map2

输入:<xxxx, hello world>                                           <xxxx, map reduce>

<xxxx, hello hadoop>                                        <xxxx, java interface>

    <xxxx, abc qaz>                                               <xxxx, java hdfs>

    <xxxx, java jvm>                                              <xxxx, spark storm>

切分:<hello, 1>                                                           <map, 1>

<word, 1>                                                           <reduce, 1>

   <hello, 1>                                                           <java, 1>

   <hadoop, 1>                                                       <interface, 1>

   <abc, 1>                                                             <java, 1>

   <qaz, 1>                                                             <hdfs, 1>

   <java, 1>                                                            <spark, 1>

   <jvm, 1>                                                             <storm, 1>

2、shuffle阶段

切分完毕后,每一组<key, value>都会不断地被collect到一个内存缓冲区中,对应代码中的数据结构MapOutputBuffer。

partition过程:每一组<key, value>在被收集的时候,就已经确定了分区(partition),即在这个时候就已经确定了要交给哪个reduce任务处理。分区会给<key, value>加上一个索引标识。假设分区后(分区算法可以设定,默认是hash值模运算),数据如下:reduce1的标识是0,reduce2的标识是1

<hello, 1>                0                                          <map, 1>                         0

<word, 1>                1                                          <reduce, 1>                      1

                                              <hello, 1>                0                                          <java, 1>                          0

                                              <hadoop, 1>             1                                          <interface, 1>                   1

                                              <abc, 1>                  0                                           <java, 1>                         0

                                              <qaz, 1>                  1                                           <hdfs, 1>                         1

                                              <java, 1>                 0                                           <spark, 1>                        0

                                              <jvm, 1>                  1                                           <storm, 1>                       1

spill过程:缓冲区默认是100M,每当里面的数据达到80M(比例80%,这个比例也可以人为设置),就会另起一个线程SpillThread往磁盘溢写,每次溢写都会产生一个文件。

sort过程:在溢写的过程中一直在排序,比较算法可以定制,默认排序算法是快速排序(可以人为设定),排序的过程就是一些位置的索引在不断的变化。

排序之后的数据:

<abc, 1>                 0                                          <hdfs, 1>                         1

<hello, 1>                0                                          <interface, 1>                   1

<hello, 1>                0                                          <java, 1>                          0

<hadoop, 1>             1                                         <java, 1>                         0

                                             <java, 1>                 0                                          <map, 1>                         0

                                             <jvm, 1>                  1                                         <reduce, 1>                      1

                                             <qaz, 1>                  1                                          <spark, 1>                        0

                                             <word, 1>                1                                          <storm, 1>                       1  

combine过程:这个过程默认是没有的,需要明确指定combiner。combiner其实就是一个reducer,可以让数据交给reduce任务之前,进行一些计算、合并。它的意义在于,使数据进一步减少,减轻了                       reduce任务通过网络获取数据的压力和reduce处理数据的压力。combiner也可以自己定制,每个溢写文件都会combine。

combiner会通过一个比较器对key进行比较,相同的key(比较结果为0,比较算法可以定制),会被放到一个集合的迭代器中,然后迭代进行一次reduce运算,产生一个输出。

combine之后的数据:

<abc, 1>                 0                                         <hdfs, 1>                        1

<hello, 1+1>             0                                        <interface, 1>                   1

<hadoop, 1>             1                                         <java, 1+1>                     0

                                             <java, 1>                 0                                          <map, 1>                         0

                                             <jvm, 1>                  1                                         <reduce, 1>                      1

                                             <qaz, 1>                  1                                          <spark, 1>                        0

                                             <word, 1>                1                                          <storm, 1>                       1

merge过程:一个map所有的溢写文件都会进行合并,产生一个最终的溢写文件和一个索引文件。

copy数据过程:每个reduce任务会远程copy属于自己的多个map输出数据文件,通过http传输,在本地会合并。另外,这个过程也会进行combine,此次不过多说明。

结果如下:

reduce0                        reduce1

<abc, 1>                     <hadoop, 1>

                      <hello, 2>                    <jvm, 1>

<java, 1>                    <qaz, 1>

                      <java, 2>                     <word, 1>

<map, 1>                     <hdfs, 1>

<spark, 1>                   <interface, 1>

<reduce, 1>

<storm, 1>

sort过程:对上述结果进行排序,结果如下:

reduce0                        reduce1

<abc, 1>                     <hadoop, 1>

                       <hello, 2>                    <hdfs, 1>

<java, 1>                    <interface, 1>

                        <java, 2>                     <jvm, 1>

<map, 1>                     <qaz, 1>

<spark, 1>                   <reduce, 1>

<storm, 1>

<word, 1>

3、reduce阶段

通过一个GroupComparator对key进行比较,相同的key(比较结果为0,比较算法可以定制),会被放到一个集合的迭代器中,然后迭代进行一次reduce运算,产生一个输出。类似combine过程。

最终的输出:                     reduce0                        reduce1

<abc, 1>                     <hadoop, 1>

                       <hello, 2>                    <hdfs, 1>

<java, 3>                    <interface, 1>

                         <map, 1>                    <jvm, 1>

<spark, 1>                   <qaz, 1>

<reduce, 1>

<storm, 1>

<word, 1>

从上述过程的分析可以看出,合并和排序是核心!!!

PS:其实每个阶段没有这么分明,只不过是为了分析和理解的需要,才进行这样详细的划分,而且划分的还不一定正确,请大家及时纠错。另外,上述流程中涉及到好多的细节,没有一一说明。

时间: 2024-10-12 23:21:48

MapReduce运行流程分析的相关文章

Struts运行流程分析与声明式验证

strust2运行流程分析 1.发送一个HttpServletRequest给StrutsPrepareAndExecuteFilter 2.StrutsPrepareAndExecuteFilter询问ActionMapper:该请求是否是一个Struts2请求(即是否返回一个非空的ActionMapping对象) 3.若ActionMapper认为该请求是一个Struts2请求,则StrutsPrepareAndExecuteFilter把请求的处理交给ActionProxy 4.Actio

thttpd和cgilua安装与运行流程分析

安装 参考如下博文安装thttpd软件 http://blog.csdn.net/21aspnet/article/details/7045845 http://blog.csdn.net/dragoncheng/article/details/5614559 thttpd配置文件: [email protected]:/usr/local/bin# cat /usr/local/thttpd/conf/ etc/  logs/ man/  sbin/ www/  [email protecte

springmvc的运行流程分析

前几篇文章对springmvc讲解的很清楚,大家看下,有问题,我们再一起讨论. 其实springmyuxvc最为重要是它的运行流程,接着,我们来分析一下,其运行过程,废话不多说,看图说话: 分析如下: 1,用户发起请求到前端控制器(DispatchService) 2,前端控制器通过HandlerMapping找到Handler(即是Action) 3,HandlerMapping返回HandlerExecutionChain(执行链),该执行链包含两部分内容,(Handler对象,拦截器数组)

SparkSteaming运行流程分析以及CheckPoint操作

本文主要通过源码来了解SparkStreaming程序从任务生成到任务完成整个执行流程以及中间伴随的checkpoint操作 注:下面源码只贴出跟分析内容有关的代码,其他省略 1 分析流程 应用程序入口: val sparkConf = new SparkConf().setAppName("SparkStreaming") val sc = new SparkContext(sparkConf) val ssc = new StreamingContext(sc, Seconds(b

yii框架详解 之 CWebApplication 运行流程分析

在 程序入口处,index.php 用一句 Yii::createWebApplication($config)->run();  开始了app的运行. 那么,首先查看 CWebApplication 的 构造函数,如下: public function __construct($config=null) { Yii::setApplication($this); // set basePath at early as possible to avoid trouble if(is_string

Hadoop Mapreduce运行流程

Mapreduce的运算过程为两个阶段: 第一个阶段的map task相互独立,完全并行: 第二个阶段的reduce task也是相互独立,但依赖于上一阶段所有map task并发实例的输出: 这些task任务分布在多台机器运行,它的运行管理是有一个master负责,这个master由yarn负责启动,那么yarn如何知道启动多少个map task进程去计算呢? 下面概述一下Mapreduce的执行流程: 1.客户端首先会访问hdfs的namenode获取待处理数据的信息(文件数及文件大小),形

mapreduce运行流程总结

先上图,下图描绘了一个mapreduce程序的的一般运行过程和需要经过的几个阶段 大体上我们可以将mapreduce程序划分为inputformat ,map ,shuffle,reduce,outputformat五个阶段,下面我们会详细介绍各个阶段的具体的运行细节 以最简单的wordcount程序为例,本例使用基于hadoop2.6的环境,一般的api都使用mapreudce下的,注意不要使用mapred下的api可能会引起未知错误  惯例hello word程序 driver类,负责构建m

[原创]java WEB学习笔记70:Struts2 学习之路-- struts2拦截器源码分析,运行流程

本博客的目的:①总结自己的学习过程,相当于学习笔记 ②将自己的经验分享给大家,相互学习,互相交流,不可商用 内容难免出现问题,欢迎指正,交流,探讨,可以留言,也可以通过以下方式联系. 本人互联网技术爱好者,互联网技术发烧友 微博:伊直都在0221 QQ:951226918 -----------------------------------------------------------------------------------------------------------------

016_笼统概述MapReduce执行流程结合wordcount程序

一.map任务处理 1 .读取输入文件内容,解析成key.value对.对输入文件的每一行,解析成key.value对.每一个键值对调用一次map函数. 2 .写自己的逻辑,对输入的key.value处理,转换成新的key.value输出.3. 对输出的key.value进行分区.4 .对不同分区的数据,按照key进行排序.分组.相同key的value放到一个集合中.5 .(可选)分组后的数据进行归约. 二.reduce任务处理 1.对多个map任务的输出,按照不同的分区,通过网络copy到不同