五、分布式并行编程模型MapReduce

1. 概述

1.1 分布式并行编程

• 背景：摩尔定律已经开始逐渐失效，提升数据处理计算能力刻不容缓。
• 传统的程序开发与分布式并行编程
  • 传统的程序开发：以单指令、单数据流的方式顺序执行，虽然这种方式比较符合人类的思维习惯，但是，这种程序的性能受到单台机器性能的限制，可扩展性较差。
  • 分布式并行编程：分布式程序运行在大规模计算机集群上，集群中包括大量廉价服务器，可以并行执行大规模数据处理任务，从而获得海量的计算能力。

1.2 MapReduce模型简介

MapReduce将复杂的、运行于大规模集群上的并行计算过程高度抽象到了两个函数：Map和Reduce，这两个函数及其核心思想都源自函数式编程语言。

• 设计理念：“计算向数据靠拢”，而不是“数据向计算靠拢"
  • 原因：数据需要大量的网络传输开销，尤其是在大规模数据环境下，这种开销尤为惊人，所以，移动计算要比移动数据更加经济
  • 措施：只要有可能，一个集群中的MapReduce框架就会将Map程序就近地在HDFS数据所在的节点运行，即将计算节点和存储节点放在一起运行，从而减少了节点间的数据移动开销。

• 架构：Master/Slave架构（一个Master和若干个Slave）

  • Master上运行JobTracker（JobTracker负责作业和任务的调度，监控它们的执行，并重新调度已经失败的任务）
  • Slave上运行 TaskTracker（TaskTracker负责执行由JobTracker指派的任务）

• 在一个MapReduce的作业中必定会涉及到如下一些组件：

  • 客户端：提交MapReduce作业
  • yarn资源管理器：负责集群上计算资源的协调
  • yarn节点管理器：负责启动和监控集群中机器上的计算容器（container）
  • MapReduce的application master：负责协调运行MapReduce的作业
  • HDFS：分布式文件系统，负责与其他实体共享作业文件

1.3 Map和Reduce函数

• 都是以<key, value>作为输入，按一定的映射规则转换成另一个或一批<key, value>进行输出。

函数	输入	输出	说明
Map	<k1,v1> 如：<行号,”a b c”>	List(<k2,v2>) 如：<“a”,1>	1、将小数据集进一步解析成一批<key,value>对，输入Map函数中进行处理 2、每一个输入的<k1,v1>会输出一批<k2,v2>。<k2,v2>是计算的中间结果
Reduce	<k2,List(v2)> 如：<“a”,<1,1,1>>	<k3,v3> 如：<“a”,3>	输入的中间结果<k2,List(v2)>中的List(v2)表示是一批属于同一个k2的value

• map:一个map函数本质上是将一种操作进行进行映射，针对不同的对象进行同一种操作
• reduce：将所得的中间结果进行混合

2. MapReduce的工作流程

2.1 工作流程概述

• 大规模数据集的处理包括分布式存储和分布式计算两个核心环节。
• 谷歌公司用分布式文件系统GFS实现分布式数据存储，用MapReduce实现分布式计算，而Hadoop则使用分布式文件系统HDFS实现分布式数据存储，用Hadoop MapReduce实现分布式计算。
• MapReduce核心思想：分而治之（与递归的思想不谋而合）
  • 即把一个大的数据集拆分成多个小数据块在多台机器上并行处理
    • **首先会被拆分成许多个Map任务在多台机器上并行执行，**每个Map任务通常运行在数据存储的节点上，这样，计算和数据就可以放在一起运行，不需要额外的数据传输开销。当Map任务结束后，会生成以<key,value>形式表示的许多中间结果。
    • 然后，这些中间结果会被分发到多个Reduce任务在多台机器上并行执行，具有相同key的<key,value>会被发送到同一个Reduce任务那里，Reduce任务会对中间结果进行汇总计算得到最后结果，并输出到分布式文件系统中。

• 注意：
  • 不同的Map任务之间不会进行通信，不同的Reduce任务之间也不会发生任何信息交换；用户不能显式地从一台机器向另一台继机器发送消息，所有的数据交换都是通过MapReduce框架自身去实现的。（通信只会在相同的map、reduce任务之间进行）
  • 在MapReduce的整个执行过程中，Map任务的输入文件、Reduce任务的处理结果都是保存在分布式文件系统中的，而Map任务处理得到的中间结果则保存在本地存储（如磁盘）中。

2.2 MapReduce的各个执行阶段

1. MapReduce框架使用InputFormat模块做Map前的预处理，比如，验证输入的格式是否符合输入定义；然后，将输入文件切分为逻辑上的多个InputSplit，InputSplit是MapReduce对文件进行处理和运算的输入单位，只是一个逻辑概念，每个InputSplit并没有对文件进行实际切割，只是记录了要处理的数据的位置和长度。
2. 因为InputSplit是逻辑切分而非物理切分，所以，还需要通过RecordReader（RR）并根据InputSplit中的信息来处理InputSplit中的具体记录，加载数据并转换为适合Map任务读取的键值对，输入给Map任务。
3. Map任务会根据用户自定义的映射规则，输出一系列的<key,value>作为中间结果。
4. 为了让Reduce可以并行处理Map的结果，需要对Map的输出进行一定的分区、排序（Sort）、合并（Combine）和归并（Merge）等操作，得到<key,value-list>形式的中间结果，再交给对应的Reduce程序进行处理，这个过程称为Shuffle。
5. Reduce以一系列<key,value-list>中间结果作为输入，执行用户定义的逻辑，输出结果给OutputFormat模块。
6. OutputFormat模块会验证输出目录是否已经存在，以及输出结果类型是否符合配置文件中的配置类型，如果都满足，就输出Reduce的结果到分布式文件系统。

2.3 Shuffle过程详解

2.3.1 Shuffle过程简介

• 所谓Shuffle，是指针对Map输出结果进行分区、排序和合并等处理，并交给Reduce的过程。因此，Shuffle过程分为Map端的操作和Reduce端的操作。

1. 在Map端的Shuffle过程。Map的输出结果首先被写入缓存，当缓存满时，就启动溢写操作，把缓存中的数据写入磁盘文件，并清空缓存。当启动溢写操作时，首先需要把缓存中的数据进行分区，然后对每个分区的数据进行排序（Sort）和合并（Combine），之后再写入磁盘文件。每次溢写操作会生成一个新的磁盘文件，随着Map任务的执行，磁盘中就会生成多个溢写文件。在Map任务全部结束之前，这些溢写文件会被归并（Merge）成一个大的磁盘文件，然后，通知相应的Reduce任务来领取属于自己需要处理的数据。

   

2. 在Reduce端的Shuffle过程。Reduce任务从Map端的不同Map机器领回属于自己需要处理的那部分数据，然后，对数据进行归并（Merge）后交给Reduce处理。

2.3.2 Map端得Shuffle过程

1. 输入数据和执行Map任务
     Map任务的输入数据一般保存在分布式文件系统（如GFS或HDFS）的文件块中，这些文件块的格式是任意的，可以是文档，也可以是二进制格式的。Map任务接受<key,value>作为输入后，按一定的映射规则转换成一批<key,value>进行输出。
2. 写入缓存
     每个Map任务都会被分配一个缓存，Map的输出结果不是立即写入磁盘，而是首先写入缓存。在缓存中积累一定数量的Map输出结果以后，再一次性批量写入磁盘，这样可以大大减少对磁盘I/O的影响。因为，磁盘包含机械部件，它是通过磁头移动和盘片的转动来寻址定位数据的，每次寻址的开销很大，如果每个Map输出结果都直接写入磁盘，会引入很多次寻址开销，而一次性批量写入，就只需要一次寻址，连续写入，大大降低了开销。需要注意的是，在写入缓存之前，key与value值都会被序列化成字节数组。
3. 溢写（分区、排序和合并）
     提供给MapReduce的缓存的容量是有限的，默认大小是100MB。随着Map任务的执行，缓存中Map结果的数量会不断增加，很快就会占满整个缓存，这时，就必须启动溢写（Spill）操作，把缓存中的内容一次性写入磁盘，并清空缓存。溢写的过程通常是由另外一个单独的后台线程来完成的，不会影响Map结果往缓存写入。但是，为了保证Map结果能够不停地持续写入缓存，不受溢写过程的影响，就必须让缓存中一直有可用的空间，不能等到全部占满才启动溢写过程，所以，一般会设置一个溢写比例，如0.8，也就是说，当100MB大小的缓存被填满80MB数据时，就启动溢写过程，把已经写入的80MB数据写入磁盘，剩余20MB空间供Map结果继续写入。
     但是，在溢写到磁盘之前，缓存中的数据首先会被分区（Partition）。缓存中的数据是<key,value>形式的键值对，这些键值对最终需要交给不同的Reduce任务进行并行处理。MapReduce通过Partitioner接口对这些键值对进行分区，默认采用的分区方式是采用Hash函数对key进行哈希后，再用Reduce任务的数量进行取模，可以表示成hash(key) mod R。其中，R表示Reduce任务的数量，这样，就可以把Map输出结果均匀地分配给这R个Reduce任务去并行处理了。当然，MapReduce也允许用户通过重载Partitioner接口来自定义分区方式。
     对于每个分区内的所有键值对，后台线程会根据key对它们进行内存排序（Sort），排序是MapReduce的默认操作。排序结束后，还包含一个可选的合并（Combine）操作。如果用户事先没有定义Combiner函数，就不用进行合并操作。如果用户事先定义了Combiner函数，则这个时候会执行合并操作，从而减少需要溢写到磁盘的数据量。
     所谓**“合并”，是指将那些具有相同key的<key,value>的value加起来，比如，有两个键值对<"xmu",1>和<"xmu",1>，经过合并操作以后就可以得到一个键值对<"xmu",2>，减少了键值对的数量。这里需要注意，Map端的这种合并操作，其实和Reduce的功能相似，但是，由于这个操作发生在Map端，所以，我们只能称之为“合并”，从而有别于Reduce。不过，并非所有场合都可以使用Combiner，因为，Combiner的输出是Reduce任务的输入，Combiner绝不能改变Reduce任务最终的计算结果，一般而言，累加、最大值等场景可以使用合并操作。
     经过分区、排序以及可能发生的合并操作之后，这些缓存中的键值对就可以被写入磁盘，并清空缓存。每次溢写操作都会在磁盘中生成一个新的溢写文件，写入溢写文件中的所有键值对，都是经过分区和排序**的。
4. 文件归并
     每次溢写操作都会在磁盘中生成一个新的溢写文件，随着MapReduce任务的进行，磁盘中的溢写文件数量会越来越多。当然，如果Map输出结果很少，磁盘上只会存在一个溢写文件，但是，通常都会存在多个溢写文件。最终，在Map任务全部结束之前，系统会对所有溢写文件中的数据进行归并（Merge），生成一个大的溢写文件，这个大的溢写文件中的所有键值对，也是经过分区和排序的。
     所谓归并（Merge)，是指对于具有相同key的键值对，会被归并成一个新的键值对。具体而言，对于若干个具有相同key的键值对<k1,v1>、<k1,v2>…，会被归并成一个新的键值对<k1,<V1,V2,...vn>>。
     另外，进行文件归并时，如果磁盘中已经生成的溢写文件的数量超过参数min.num.spills.for.combine的值时（默认值是3，用户可以修改这个值）。那么，就可以再次运行Combiner，对数据进行合并操作，从而减少写入磁盘的数据量。但是，如果磁盘中只有一两个溢写文件时，执行合并操作就会“得不偿失”，因为执行合并操作本身也需要代价，因此，不会运行Combiner。

2.3.3 Reduce端得Shuffle过程

Reduce端的Shuffle过程非常简单，只需要从Map端读取结果，然后执行归并操作，最后输送给Reduce任务进行处理，具体执行流程如下：

1. “领取”数据
     Map端的Shuffle过程结束后，所有Map输出结果都保存在Map机器的本地磁盘上，Reduce任务需要把这些数据“领取”（Fetch）回来，存放到自己所在机器的本地磁盘上。因此，在每个Reduce任务真正开始之前，它大部分时间都在从Map端把属于自己处理那些分区的数据“领取”过来。
     每个Reduce任务会不断地通过RPC（Remote Procedure Call）向JobTracker询问Map任务是否已经完成；JobTracker监测到一个Map任务完成后，就会通知相关的Reduce任务来“领取”数据；一旦一个Reduce任务收到JobTracker通知，它就会到该Map任务所在机器上把属于自己处理的分区数据领取到本地磁盘中。一般系统中会存在多个Map机器，因此，Reduce任务会使用多个线程同时从多个Map机器领回数据。
2. 归并数据
     从Map端领回的数据，会首先被存放在Reduce任务所在机器的缓存中，如果缓存被占满，就会像Map端一样被溢写到磁盘中。由于在Shuffle阶段，Reduce任务还没有真正开始执行，因此，这时可以把内存的大部分空间分配给Shuffle过程作为缓存。需要注意的是，系统中一般存在多个Map机器，所以，Reduce任务会从多个Map机器领回属于自己处理的那些分区的数据，因此，缓存中的数据是来自不同的Map机器的，一般会存在很多可以合并（Combine）的键值对。
     当溢写过程启动时，具有相同key的键值对会被归并（Merge），如果用户定义了Combiner，则归并后的数据还可以执行合并操作，减少写入磁盘的数据量。每个溢写过程结束后，都会在磁盘中生成一个溢写文件，因此，磁盘上会存在多个溢写文件。最终，当所有的Map端数据都已经被领回时，和Map端类似，多个溢写文件会被归并成一个大文件，归并的时候还会对键值对进行排序，从而使得最终大文件中的键值对都是有序的。当然，在数据很少的情形下，缓存就可以存储所有数据，就不需要把数据溢写到磁盘，而是直接在内存中执行归并操作，然后直接输出给Reduce任务。
     需要说明的是，把磁盘上的多个溢写文件归并成一个大文件，可能需要执行多轮归并操作。每轮归并操作可以归并的文件数量是由参数io.sort.factor的值来控制的（默认值是10，可以修改）。
     假设磁盘中生成了50个溢写文件，每轮可以归并10个溢写文件，则需要经过5轮归并，得到5个归并后的大文件。
3. 把数据输入Reduce任务
     磁盘中经过多轮归并后得到的若干个大文件，不会继续归并成一个新的大文件，而是直接输入给Reduce任务，这样可以减少磁盘读写开销。由此，整个Shuffle过程顺利结束。接下来，Reduce任务会执行Reduce函数中定义的各种映射，输出最终结果，并保存到分布式文件系统中。

3. 以WordCount为例理解MapReduce过程

1. 判断WordCount任务是否可以采用MapReduce实现。（满足前提条件：待处理的数据集可以分解成许多小的数据集，而且每一个小数据集都可以完全并行地进行处理，核心是并行处理）本例中，不同单词之间的频数不存在相关性，彼此独立，可以把不同的单词分发给不同的机器进行并行处理，因此，可以采用MapReduce来实现词频统计任务。

2. 确定MapReduce程序的设计思路。本例中，把文件内容解析成许多个单词，然后把所有相同的单词聚集到一起。最后，计算出每个单词出现的次数进行输出

3. 确定MapReduce程序的执行过程。把一个大文件切分成许多个分片，每个分片输入给不同机器上的Map任务，并行执行完成“从文件中解析出所有单词”的任务。Map的输入采用Hadoop默认的<key, value>输入方式，即文件的行号作为key，文件的一行作为valueMap的输出以单词作为key，1作为value，即<单词，1>，表示单词出现了1次。

4. Map阶段完成后，会输出一系列<单词,1>这种形式的中间结果，然后，Shuffle阶段会对这些中间结果进行排序、分区，得到<key, value-list>的形式（比如<hadoop,<1,1,1,1,1>>），分发给不同的Reduce任务。Reduce任务接收到所有分配给自己的中间结果（一系列键值对）以后，就开始执行汇总计算工作，计算得到每个单词的频数并把结果输出到分布式文件系统。

3.1 WordCount实现过程

3.2 简易版MapReduce工作流程

3.3 数据分片

MapReduce的工作流程：

• Inputformat的作用：加载、读取HDFS中的文件，对输入进行格式验证；将大文件切分成许多分片split，但此切分仅是逻辑上的切分，即逻辑定义每个split的起点和长度，并非真正意义的物理切分。
• record reader：记录阅读器，根据split的位置和长度，从HDFS中的各个块读取相关分片，读取成<k,v>的形式。

3.4 WordCount详细讲解

1. 数据分片

   

2. split的map过程

   

3. Reduce过程

   

4. WordCount的Map过程

   

5. WordCount的Reduce过程

   

6. Shuffle过程

   

3.5 详细版MapReduce工作流程

3.6 MapReduce的体系结构

• Client（客户端）

  • 主要功能：负责提交作业，查看作业状态

  • 提交作业：用户编写的MapReduce程序通过Client提交到JobTracker端。

  • 查看作业状态：用户可通过Client提供的一些接口查看作业运行状态。

• JobTracker（作业跟踪器）

  • 主要功能：负责资源监控、作业调度

  • 资源监控：JobTracker监控所有TaskTracker与Job的健康状况，一旦发现节点失效（通信失败或节点故障），就将相应的任务转移到其他节点。

  • 作业调度：JobTracker会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器（TaskScheduler），而任务调度器会选择合适的（比较空闲）节点资源来执行任务。

• TaskScheduler（任务调度器）

  • 执行具体的相关任务，一般接收JobTracker发送过来的命令。

  • 把一些自己的资源使用情况，以及任务的运行进度通过心跳的方式，也就是heartbeat发送给JobTracker。

• TaskTracker（任务跟踪器）

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

  • TaskTracker使用slot等量划分本节点上的资源量（CPU、内存等）。一个Task获取到一个slot后才有机会运行，而Hadoop调度器（TaskScheduler）的作用就是将各个TaskTracker上的空闲slot分配给Task使用。slot分为Map slot和Reduce slot两种，分别供MapTask和Reduce Task使用。

4. 实验（之后统一完成）

参考自DataWhale学习资料