目录

MapReduce概述 

分布式并行编程

MapReduce模型简介

MapReduce体系结构

MapReduce工作流程 

MapReduce工作特点 

Shuffle过程原理

MapReduce完成的shuffle过程

MapReduce应用程序执行过程

实例分析：WordCount

MapReduce的具体应用

---

MapReduce概述 

分布式并行编程

    数据处理能力提升的两条路线

        单核CPU到双核到四核到八核

        分布式并行编程：不是把程序在一台单机上运行，借助一个集群通过多台机器同时并行处理大规模数据集

            分布式并行编程是多台计算机同时在运行相关的程序

分布式与集群是什么？区别是什么？

    MapReduce和传统的并行编程框架

	集群的架构和容错性	硬件价格及扩展性	编程和学习难度	适用场景
传统的并行编程框架	采用共享式架构（共享内存、共享底层的存储空间），例如性能计算模型HPC 1、扩展比较难2、因为硬件之间紧密耦合，所以其中一个硬件发生故障以后容易导致整个集群不可工作	采用刀片服务器，高速网络一集存储区域网络SAN 普通刀片服务器几万，好一点的十几万（成本高） 且只能纵向扩展（增加内存、更多磁盘）	编程难度高 (程序员不仅要解决要做什么，同时还要解决怎么做（必须要通过程序的方式告诉它怎么做）) 编程原理和多线程的编程逻辑比较类似，来实现不同任务之间的同步	用于实施的细粒度计算 尤其适用于计算密集型的应用
MapReduce	采用典型的非共享式架构 在整个集群当中，每个节点都拥有自己的内存（节点之间非共享），任何一个节点出现问题，不会影响其他节点正常运行，整个集群中又设计了冗余和容错机制	整个集群可以随意增加减少相关的计算节点，这个个人计算机就可以（成本低、扩展性好）	MapReduce只需要告诉系统你要做什么事情，而不需要开发人员去告诉它如何去做。它会屏蔽掉整个分布式程序运行底层的所有细节（包括同步以及不同进程之间的通信、负载均衡、失败的恢复等）整个都不需要开发人员去做，整个系统框架自动帮你实现 MapReduce会自动实现分布式部署，部署到集群上面的各个机器上面去运行	一般适用于非实时的批处理以及数据密集型的应用

 

MapReduce模型简介

    MapReduce整个框架的核心设计就是两个函数：Map和Reduce

    MapReduce的策略：采用分而治之

        当采用MapReduce去做大规模数据处理的时候

            MapReduce会把非常庞大的数据集，切分成非常多的独立的小分片（spilt）

                 然后为每一个分片单独的启动一个map任务

                      最终通过多个map任务，并行地在多个机器上去处理

 

MapReduce的理念：计算向数据靠拢而不是数据向计算靠拢

    数据向计算靠拢是指：完成一次数据分析时，选择一个计算节点，把运行数据分析的程序放到计算节点上运行

        然后把它涉及到的数据，全部从各个不同节点上面拉过来，传输到计算发生的地方

但是在大数据时代，把所有的数据拉到一个计算节点上去计算，基本上是不可能的（计算机处理起来会崩溃）

    计算向数据靠拢是指：在做应用程序计算的时候，构建一个集群，在集群中选择相关的机器作为map机器，这些map机器负责相关的数据处理分析，还有一些是Reduce机器，做Reduce任务处理。

        拿这些机器去做数据处理，假如说要去处理某一很大的数据集（数据集被分布的存储到各个节点上面（HDFS是以块为单位           存储数据），这些数据块都是存储到不同的机器节点上面去），假设某一个数据块存储在机器A上面，对于MapReduce来             说，如果想对数据块A进行分析的话，但是它又不存在机器A上面，这时，MapReduce会去寻找离机器A最近的map机器去完         成对数据块的处理。通常来讲比较理想情况下，map任务就是和数据块在同一个机器上面，这就是实现了计算向数据靠拢，         数据不需要发生迁移，计算就可以在数据节点上面执行，完成运算得到结果，最终把结果汇总到相关的任务协调节点，提交         给用户。

       通过这种方式：大大减少了整个网络当中的数据传输开销，从而提升了整个分布式程序处理性能。

 

MapReduce的架构：采用了master/slave的架构

Master/slave架构中包括：一个master服务器以及若干个slave服务器（Master就是名称节点，Slave就是数据节点）

    Master上面会运行作业跟踪器（JobTracker）

        JobTracker负责整个作业的调度和处理以及失败和恢复

slave上面会运行负责具体任务执行的组件（TaskTracker）

        TaskTracker负责接收JobTracker给它发的作业处理指令，完成具体的任务处理

 

 

 

 

MapReduce体系结构

这个图有点小错误就是Task Schedule是任务分配器，当时写的时候理解的不是很好

Client（客户端）

通过Client可以提交用户编写的应用程序，用户通过它将应用程序交到JobTracker端

通过这些Client，用户也可以通过它提供的一些接口去查看当前作业的运行状态

通常在Hadoop中，通常来讲应用程序到了集群当中都是表现为作业的形式

JobTracker（作业跟踪器（管家））

负责资源的监控和作业的调度

监控底层的其他的TaskTracker以及当前运行的Job的健康状况

一旦探测到失败的情况就把这个任务转移到其他节点继续执行跟踪任务执行进度和资源使用量，并会把这些信息发送给任务调度器Task Schedule

Task Schedule

负责任务调度（解决应该把哪个Task（任务）分发给哪个TaskTracker去执行（各个TaskTracker是分布在不同的机器上面，负责执行具体任务））

它会从具有空闲资源的节点中选出相关节点，然后把Task分配到这些节点上去执行

Task Schedule是一个可插拔模块，允许用户自己去编写调度模块，采用自己的任务调度策略去实现

TaskTracker

TaskTracker 会周期性地通过“心跳”将本节点上资源的使用情况和任务的运行进度汇报给JobTracker，同时接收JobTracker 发送过来的命令并执行相应的操作（如启动新任务、杀死任务等）

TaskTracker是以一种什么方式衡量资源使用情况？

TaskTracker使用一个槽的概念，叫slot。它会把机器上面所有的CPU内存资源进行打包，然后把资源进行等分，等分成很多slot。slot是一个资源调度单位。一个Task 获取到一个slot 后才有机会运行，而Hadoop调度器的作用就是将各个TaskTracker上的空闲slot分配给Task使用。slot 分为Map slot 和Reduce slot 两种，分别供MapTask 和Reduce Task 使用。两种类型slot是不通用的。如果有10个slot，那么5个给map、5个给reduce，如果reduce任务是空闲的，一个槽都没用，map的槽已经被map任务占满了，而且还有一个map任务等待执行，但reduce槽也不会分给排队的map任务，该缺陷Hadoop2.0中做出了修改。

 所以是以slot为单位调度资源，所以是只有这个机器上有空闲的slot，才能把相关的Task分配给他执行，如果有空闲的map slot，就把map Task分配给他执行。如果有空闲的reduce slot，就把reduce Task分配给他去执行

Task(任务)

Task包括map任务（专门执行map函数）和reduce任务（专门执行reduce函数）

这两个任务都是由TaskTracker来启动的

在同一台机器上面是可以同时运行map Task和reduce Task这两种任务的

　　

　　Map任务的数量，即谁来完成任务分配和数据分片，名称节点还是数据节点？

　　名称节点。

　　　　HDFS以固定大小的block为基本单位存取数据，而对于MapReduce而言，其处理单位是Split。Split 是一个逻辑概念，它只包含一些元数据信息，比如数据起始位置、数据长度、数据所在节点等，它的划分方法完全由用户自己决定。　但大多数情况下，理想的分片大小是一个HDFS块。（下图的分片是有问题的，split1和split4的数据分片涵盖了两个数据节点，Map任务执行时需要移动数据。）

 

MapReduce工作流程 

 

 

 分片：

InputFormat会把一个大的文件分成很多split，那split如何分呢？

一个文件被这样切分成了4片（逻辑分片），但其实它有六个块（物理块）组成，这个文件在底层是分布式存储在不同的机器上面，一共存储了六个块，但每个块都有冗余存储。

分片和块是不同的概念

当每一次产生一个分片的时候，Hadoop都会给这个分片生成一个map任务。所以，整个集群里面，分了多少个分片，就会产生多少个map任务。

如果分片过多，导致map任务启动过多，map任务之间需要进行相关的切换的时候，要耗费相关的管理资源，影响执行效率。

如果分片过少，又会影响程序的并行度，达不到提升整个程序处理效率的目的。

现在，实际应用中，会把一个块的大小作为分片的大小。例如：上图，第一个分片时候占了block1和block2，那map机器在block1上面，如果需要block2的值，还需要把block2的数据传输到block1，这就造成了额外的数据传输开销

 

map任务数量和reduce任务数量是如何确定的？

在Hadoop集群当中，执行MapReduce任务的时候，map任务数量就是由产生分片的数量来决定的（因为产生一个分片就需要启动一个map任务）。

reduce任务数量：在极端情况下，如果不考虑任何并行度，就可以只启动一个reduce任务。但通常来讲reduce任务都是在多个机器上，有多个任务同时运行，这样可以增加并行度。

reduce任务如何设置？

最优的reduce任务个数取决于集群中可用的reduce任务槽（slot）的数目。

通常设置比reduce任务槽数目稍微小一些的reduce任务个数（这样可以预留一些系统资源处理可能发生的错误）

 

　　　　一个数据节点上可能有某个文件的多个数据块。

　　　　多个数据块可以分成多个数据分片，一个数据块也可以分成多个数据分片。

　　　　一个数据分片分配给一个Map任务进行处理。

　　　　一个数据节点上可以运行多个Map任务。

　　　　

　　　　一个数据节点上能否运行多个Reduce任务？

　　　　可以，但是一般不会这么分配

 

MapReduce工作特点 

不同的Map任务之间不会进行通信，不同的Reduce任务之间也不会发生任何信息交换。用户不能显式地从一台机器向另一台机器发送消息，所有的数据交换都是通过MapReduce框架自身去实现的。Map任务的输入文件、Reduce任务的处理结果都是保存在分布式文件系统中的，而Map任务处理得到的中间结果则保存在本地存储中。只有当Map处理全部结束后，Reduce过程才能开始。只有Map需要考虑数据局部性，实现“计算向数据靠拢”，Reduce则无需考虑数据局部性

一个Map任务，能不能处理多个不同的<k1,v1>键值对？

　　　　 一个Map任务是可以处理多个不同的<k1,v1>值的，并且对于每一个输入的<k1,v1>都会输出一批<k2,v2>

　　　　

Shuffle过程原理

 

 

 

 

map任务和reduce任务可以在同一台机器上，也可以在不同的机器上。

我们先假设在不同的机器上面。

现在看起来是只有一个map任务和一个reduce任务，但实际上在集群当中有非常多个map任务以及非常多reduce任务，且很多map任务和reduce任务同时执行。

 现在只是放大了一个map任务和一个reduce任务。

数据是保存在分布式文件系统HDFS中，所以，输入，从文件式分布系统当中输入数据，输入完以后进行相关的分片处理，一个分片对应一个map任务，交给map函数，Map里面包含了用户对数据的处理逻辑，所以这个Map任务会运行用户处理逻辑以后，输出处理结果，那这个处理结果是一堆的键值对，这些键值对并不直接发送给reduce任务。

会先写到缓存当中，当缓存满的时候发生溢写，把缓存中的数据写到磁盘里面生成磁盘文件，同时把缓存清空掉，当把缓存当中的数据溢写到磁盘中的时候，会经历分区、排序、可能发生的合并以及文件的归并。

溢写发生多次，会生成多个磁盘文件，对于每一个磁盘文件，会进行统一的归并，多个磁盘文件归并为一个大的文件，归并完以后这个磁盘文件当中就包含了所有的键值对（都是经过分区、排序的），最终会得到一个大文件，它会通知相关Reduce任务把它取走。

 

 

为什么要分区？Map输出一些键值对，最终是扔给多个不同的reduce任务去处理，所以一定要分区。如果有4个reduce任务，就会分成四个分区，每个分区都由对应的reduce任务把它取走。

分区后进行排序（默认操作），把键值对根据key的字典序进行排序，排完序之后，每个分区里面的数据都是有序的。

可能发生的操作：合并（Combine）和Merge不是一个概念，要区分开。合并是为了减少溢写到磁盘当中的数据量。由两个键值对合并完之后变为一个键值对，由此，可以减少很多键值对，这样写入的数据量就会大大减少

之后写入磁盘，生成一个磁盘文件。

80MB缓存一满就把80MB写入磁盘，生成一个80MB的磁盘文件，溢写一次生成一个磁盘文件

随着溢写的多次发生，最终在磁盘上会有多个溢写文件，这多个溢写文件，最终在整个Map任务运行结束之前，系统会进行归并，放入本地磁盘。这大的文件里面的键值对都是分区的而且是排序的。

JobTracker会跟踪Map任务，JobTracker一旦检测到Map任务写完数据，JobTracker就会通知相应的Reduce任务，把这些属于该Reduce任务处理的分区数据让Reduce拉走。

 

 

 

 MapReduce完成的shuffle过程

　　

 

 

MapReduce应用程序执行过程

第一步，进行程序部署，把程序分配到不同的机器上面去，要从整个集群当中选出相关的机器，有些机器作为Master（只有一个），其他机器作为Worker。

Master机器作为一个管家的角色（上面运行JobTracker），其他机器作为Worker，Worker可以执行Map任务，也可以执行Reduce任务（在一堆的Worker中，选出一部分做Map

任务，另外一部分Worker做reduce任务）。要把执行的用户逻辑，分发到这些机器上去。这就是第一步完成了程序的部署。

第二步，集群中已经有这么多的Worker，接下来选出一部分Worker，把它作为执行map任务的机器，另一部分Worker作为执行reduce任务的机器，然后对于执行map任务的机器，因为运行map任务，所以需要输入数据，输入数据一般都会非常大的文件，所以需要对文件进行拆分，要对一个大的数据集进行分片，分成多个片段，分完片段以后，需要从集群中执行map任务的机器里面，选出几个空闲的机器，让它们执行数据分片的处理。

第三步，从分布式文件系统各个分片中把数据读入，读入以后生成相关的（key，value）键值对，然后提交给map任务去执行，Map里面包含了用户编写的应用程序处理逻辑，然后会输出一堆相关的（key，value），输出结果不是直接写入磁盘，也不是直接发送给reduce任务，而是先写到缓存。

第四步，一旦缓存里的写满以后，就要溢写到磁盘，所有要把缓存当中的数据，经过分区排序或者可能发生的合并操作，再把它写到磁盘当中去，写到磁盘以后就生成一个大的文件，这个文件当中包含的是很多个分区的数据，然后这个分区里的数据，都是排序的合并以后的数据，这些数据要发送给远端的reduce任务去处理。

第五步，负责执行相关reduce任务的机器，会从各个相关的map任务机器，把属于自己处理的数据都拉回去，拉回到本地进行相关的处理，处理完以后执行用户自定义的reduce函数，完成数据处理，处理结果就是一堆的键值对。

第六步，把相关的结果写到输出文件当中去，这个输出文件会写入分布式文件系统HDFS。

这是MapReduce应用程序基本执行过程，包含五大阶段，输入文件、map阶段、中间文件阶段、reduce阶段和输出文件阶段。

中间过程数据是不写入到HDFS中的，而是写入磁盘中，map任务在那个机器上边，中间结果就写到那个机器磁盘上面

 

实例分析：WordCount

 

 

只有特定的任务，能满足分而治之的任务，才可以使用MapReduce去解决。

可以把一个大的块切分成很多小的块，这些小的块分别放在不同机器上并行执行，大家互相并行执行，彼此结果不会依赖对方的结果，这个是可以满足分而治之的策略的数据集。

如果说数据集之间的结果互相依赖，它的结果依赖另一个结果，那就不可以，那就不能用MapReduce去做。

 

MapReduce的具体应用

　　具体应用：自然连接