zoukankan      html  css  js  c++  java
  • 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所有的溢写文件都会进行合并,产生一个最终的溢写文件和一个索引文件。合并是针对于不同的溢写文件中相同分区的数据。在这个合并的过程中,也会进行combine操作(如果设置了的话),此处的combine过程同上,不再细说。

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

  • 相关阅读:
    Android优化之软引用和弱引用
    Android中Service与IntentService的使用比较
    Android配置文件,所有权限
    为什么Android手机总是越用越慢?
    Android之安全机制
    JDK8-函数式接口
    JDK8-Java Streams 收集器
    JDK8- java.util.stream 库笔记
    MVVM核心实现代码(简易实现)
    JavaScript小总结
  • 原文地址:https://www.cnblogs.com/cz123/p/5447569.html
Copyright © 2011-2022 走看看