数分-理论-大数据5-MapReduce（分布式并行编程模型）

(数据分析系列)

1知识点

• 分布式并行编程
• 简介
• 工作流程
• 执行阶段
• shuffle过程详解
• WordCount理解MapReduce

2具体内容

2.1分布式并行编程

• 分布式并行编程-》提高程序性能
• 集群：大量廉价服务器，并行执行大规模数据处理任务，获取海量计算能力
• 分布式并行程序可以运行在由大量计算机构成的集群上，从而可以充分利用集群的并行处理能力，同时，通过向集群中增加新的计算节点，就可以很容易实现集群计算能力的扩充。
  

2.2简介

MapReduce:计算向数据靠拢

• 数据需要大量的网络传输开销
• 移动计算比移动数据更经济
• 集群中的MapReduce框架就会将Map程序就近地在HDFS数据所在的节点运行，即将计算节点和存储节点放在一起运行，从而减少了节点间的数据移动开销
• Master/Slave架构，一个Master多个Slave，Master上运行JobTracker，Slave上运行TaskTracker。用户提交每个计算作业，会被划分成若干任务
  • JobTracker，作业和任务调度，监控执行，并重新调度已经失败的任务
  • TaskTracker，执行 JobTracker指派的任务
• 批量计算的框架，包括，从数据输入、数据处理（map、reduce、combiner等 ）、数据输出
• 组件：
  • 客户端：提交MapReduce作业
  • yarn资源管理器：集群上计算资源的协调
  • yarn节点管理器：启动和监控集群中集群上的计算容器（container）
  • application master：协调运行MapReduce作业
  • HDFS：分布式文件系统，与其他实体共享作业文件

2.3 MapReduce函数

函数	输入	输出	说明
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

简单WordCount实例

• 统计一个文本文件中每个单词出现的次数
• Map函数自己写:
  • 输入： <某一行文本在文件中的偏移位置，该行文本的内容>
  • 输出：文件中的一行读取后解析出每个单词，输出一批中间结果<单词，出现次数>
• Reduce函数自己写:
  • 中间结果作为Reduce函数的输入
  • 用户可以将相同单词的出现次数进行累加，输出每个单词出现的总次数

2.4工作流程

• 分布式存储、分布式计算

  • 分布式文件系统HDFS实现分布式数据存储
  • MapReduce实现分布式计算
  • 输入和输出都需要借助于分布式文件系统进行存储，这些文件被分布存储到集群中的多个节点上
  • “分而治之”：把一个大的数据集拆分成多个小数据块在多台机器上并行处理

• 工作流程：

• Map：

  • MapReduce作业拆成多个Map任务，在多台机器上并行执行
  • 运行在数据存储节点（计算存储一起没有数据传输开销）
  • 结果：生成<k,v>中间结果

• Reduce：

  • 中间结果被分到多个Reduce任务
  • 在多台机器上并行执行，具有相同key的<k,v>被发送到同一个Reduce里
  • Reduce对中间结果进行汇总计算得到最后结果，并输出到分布式文件系统
    

• 不同Map任务间不通信，不信息交换

• 所有信息通过MapReduce框架自身实现

• Map任务的输入文件、Reduce任务的处理结果都是保存在分布式文件系统（HDFS）中的，而Map任务处理得到的中间结果则保存在本地存储（如磁盘）中。

2.5执行阶段

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.6shuffle过程详解

Shuffle是MapReduce的核心

2.6.1过程

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

1.Map

• Map结果写入缓存
• 缓存满时，就启动溢写操作，把缓存中的数据写入磁盘文件，并清空缓存
• 溢写操作，把缓存中的数据进行分区，然后对每个分区的数据进行排序（Sort）和合并（Combine），之后再写入磁盘文件
• 每次溢写操作生成一个新磁盘文件。Map全部结束钱，溢写文件被归并（Merge）成一个大的磁盘文件，通知相应的Reduce来领取属于自己需要处理的数据
  

2.Reduce

• 从Map端不同Map机器领回属于自己需要处理的数据
• 对数据进行归并（Merge）后交给Reduce处理

2.6.2Map

1.数据数据和执行Map任务

• 文件在文件系统中（GFS、HDFS），格式任意（文档、二进制）
• 接受<k,v>输入后，按映射规则转换成一批<k,v>输出。

2.写入缓存

• Map结果先写入缓存：
• 积累一定数量后一次性批量入磁盘，减少对磁盘I/O影响，一次寻址连续写入降低开销
• k，v被序列化成字节数组

3.溢写（分区、排序和合并）

• 默认缓存容量有限，100M
• 溢写操作（spill），把缓存中的内容一次性写入磁盘，清空缓存
  • 溢写，另一个单独的后台线程完成，不影响Map结果写入缓存
  • 为了保证Map不停持续写入，设置溢写比例。0.8：当100MB大小的缓存被填满80MB数据时，就启动溢写过程
• 分区（partition），缓存中数据<k,v>交给不同的Reduce并行处理。
  • 通过Partitioner接口对这些键值对进行分区
  • 默认分区：Hash对Key哈希，用Reduce任务数量取模。hash(key) mod R
  • R：Reduce任务的数量
  • 允许用户通过重载Partitioner接口来自定义分区方式
• 内存排序（sort），根据key对每个分区内的所有键值对排序
• 合并（combine）：事先定义了Combiner函数，则这个时候会执行合并操作，从而减少需要溢写到磁盘的数据量
  • 将具有相同key的<k，v>的v加起来：<“xmu”,1>和<“xmu”,1>，经过合并操作以后就可以得到一个键值对<“xmu”,2>
  • 减少了键值对的数量
  • Map端的这种合并操作，其实和Reduce的功能相似，称为“合并”
  • Combiner非必须
  • Combiner的输出是Reduce任务的输入，Combiner绝不能改变Reduce任务最终的计算结果
  • 一般而言，累加、最大值等场景可以使用合并操作
• 经过分区、排序、可能的合并，缓存中的键值对被写入磁盘，并清空缓存。
• 每次溢写会在磁盘生成一个新的溢写文件，写入溢写文件中的键值对都经过分区，排序

4.文件归并（Merge）

• 对具有相同key的键值对，被归并成一个新的键值对
• 具有相同key的键值对：<k1,v1>、<k1,v2>…，会被归并成一个新的键值对<k1,<V1,V2,…vn>>
• 已生成的溢写文件的数量超过参数min.num.spills.for.combine的值时（默认值是3，用户可以修改这个值）。那么，就可以再次运行Combiner，对数据进行合并操作，从而减少写入磁盘的数据量。
• 溢写文件少时不允许combiner

2.6.3Reduce

Map端读取结果，归并，送给Reduce处理
1.“领取”数据(Fetch)

• 从Map机本地磁盘“领取”数据，存在自己机器磁盘
• 每个reduce任务通过RPC（Remote Procedure Call），向JobTracker询问Map任务是否完成
• JobTracker监测到一个Map任务完成后，就会通知相关的Reduce任务来“领取”数据
• Reduce任务收到JobTracker通知，它就会到该Map任务所在机器上把属于自己处理的分区数据领取到本地磁盘中
• 系统中会存在多个Map机器，因此，Reduce任务会使用多个线程同时从多个Map机器领回数据。

2.归并数据

• Shuffle阶段，Reduce任务未真正开始，内存大部分分给shuffle作为缓存
  • 系统存在多个Map机器，Reduce任务从多个Map机器领回自己的数据
  • 一般会合并（combine）键值对
• 溢写启动，相同key归并。
  • 定义了Combiner，归并后的数据还可执行合并，减少写入磁盘的数据量
  • 多个溢写文件归并成一个大文件，并对键值对排序
  • 缓存足够时，不溢写到磁盘，直接内存中归并，输出Reduce任务
  • 多个溢写文件归并成一个大文件，需多轮归并操作。每轮归并操作可以归并的文件数量是由参数io.sort.factor的值来控制的（默认值是10，可以修改），归并一轮一个大文件

3.把数据输入Reduce任务

• 多轮归并后得到的若干个大文件，不会继续归并成一个新的大文件，而是直接输入给Reduce任务（减少磁盘读写开销）
• Reduce任务会执行Reduce函数中定义的各种映射，输出最终结果，并保存到分布式文件系统中。
  

2.7以WordCount例子，理解MapReduce

MapReduce使用前提：待处理的数据集可以分解成许多小的数据集，而且每一个小数据集都可以完全并行地进行处理。

WordCount任务：

1. 判断是否可以MapReduce实现：不同单词间的频数不存在相关性，彼此独立

   • 不同单词分发给不同机器并行处理
   • MapReduce实现词频统计任务

2. 设计思路：文件内容解析成许多个单词，把所有相同的单词聚集到一起

   • 计算出每个单词出现的次数

3. 执行过程：把一个大文件切分成许多个分片，每个分片输入给不同机器上的Map任务，并行执行完成“从文件中解析出所有单词”任务。

   • Map的输入采用Hadoop默认的<k,v>输入
     • 文件行号key
     • 文件一行为valueMap的输出以单词为key，1为value，<word,1>，单词出现了1次
   • Shuffle对中间结果排序、分区，得到<key,value-list>，分给Reduce任务
   • Reduce接到分配的中间结果（键值对），执行汇总计算，并把频数结果输出到分布式文件系统

2.7.1wordcount过程图

2.7.2工作流程

2.7.3数据分片

Map：

• Inputformat
  • 加载、读取HDFS文件，对输入格式验证
  • 大文件逻辑分成许多分片split，定义每个split起点和长度
• record reader
  • 记录阅读器，根据split位置和长度，从HDFS各个块读取相关分片，<k,v>

2.7.4过程详解

1. 数据分片
   
2. split的map过程
   
3. reduce
   
4. wordcount的map
   
5. wordcount的reduce
   
6. shuffle
   

2.7.5MapReduce工作过程

2.7.6MapReduce体系结构

2.7.6.1client客户端

• 负责提交作业，查看作业状态
• 提交作业：MapReduce程序通过client提交到JobTracker端
• 查看作业状态：通过Client提供的接口查看作业运行状态

2.7.6.2JobTracker作业跟踪器

• 负责资源监控、作业调度
• 资源监控：JobTracker监控所有TaskTracker与Job的健康状况，一旦发现节点失效（通信失败或节点故障），就将相应的任务转移到其他节点。
• 作业调度：JobTracker会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器（TaskScheduler），而TaskScheduler会选择合适的（比较空闲）节点资源来执行任务。

2.7.6.3TaskScheduler任务调度器

• 接受JobTracker发送命令，执行具体任务
• 发送heartbeat（资源使用情况，任务运行进度）给JobTracker

2.7.6.4TaskTracker（任务跟踪器）

• 周期性根据“心跳”，将本节点资源使用情况和任务运行进度汇报给JobTracker，同时接受JobTracker发送命令，执行相应操作（启动信任我，杀死任务等）
• 使用slot等量划分本节点资源（cpu，内存）
  • 一个task获取一个slot才有机会运行
  • TaskScheduler将各个TaskTracker上的空闲slot分配给Task使用
  • Map slot供MapTask使用
  • Reduce slot供ReduceTask使用

3参考

• https://shenhao-stu.github.io/Big-Data/#/