罗湖附近公司做网站建设哪家服务周到,互联网开发工程师证书,如何去掉网站后缀wordpress,网站开发源程序五、分布式并行编程模型MapReduce
1. 概述
1.1 分布式并行编程
背景#xff1a;摩尔定律已经开始逐渐失效#xff0c;提升数据处理计算能力刻不容缓。传统的程序开发与分布式并行编程 传统的程序开发#xff1a;以单指令、单数据流的方式顺序执行#xff0c;虽然这种方式…五、分布式并行编程模型MapReduce
1. 概述
1.1 分布式并行编程
背景摩尔定律已经开始逐渐失效提升数据处理计算能力刻不容缓。传统的程序开发与分布式并行编程 传统的程序开发以单指令、单数据流的方式顺序执行虽然这种方式比较符合人类的思维习惯但是这种程序的性能受到单台机器性能的限制可扩展性较差。分布式并行编程分布式程序运行在大规模计算机集群上集群中包括大量廉价服务器可以并行执行大规模数据处理任务从而获得海量的计算能力。
1.2 MapReduce模型简介
MapReduce将复杂的、运行于大规模集群上的并行计算过程高度抽象到了两个函数Map和Reduce这两个函数及其核心思想都源自函数式编程语言。
设计理念“计算向数据靠拢”而不是“数据向计算靠拢 原因数据需要大量的网络传输开销尤其是在大规模数据环境下这种开销尤为惊人所以移动计算要比移动数据更加经济措施只要有可能一个集群中的MapReduce框架就会将Map程序就近地在HDFS数据所在的节点运行即将计算节点和存储节点放在一起运行从而减少了节点间的数据移动开销。 架构Master/Slave架构一个Master和若干个Slave Master上运行JobTrackerJobTracker负责作业和任务的调度监控它们的执行并重新调度已经失败的任务Slave上运行 TaskTrackerTaskTracker负责执行由JobTracker指派的任务 在一个MapReduce的作业中必定会涉及到如下一些组件 客户端提交MapReduce作业yarn资源管理器负责集群上计算资源的协调yarn节点管理器负责启动和监控集群中机器上的计算容器containerMapReduce的application master负责协调运行MapReduce的作业HDFS分布式文件系统负责与其他实体共享作业文件
1.3 Map和Reduce函数
都是以key, value作为输入按一定的映射规则转换成另一个或一批key, value进行输出。
函数输入输出说明Mapk1,v1 如行号,”a b c”List(k2,v2) 如“a”,11、将小数据集进一步解析成一批key,value对输入Map函数中进行处理 2、每一个输入的k1,v1会输出一批k2,v2。k2,v2是计算的中间结果Reducek2,List(v2) 如“a”,1,1,1k3,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的各个执行阶段
MapReduce框架使用InputFormat模块做Map前的预处理比如验证输入的格式是否符合输入定义然后将输入文件切分为逻辑上的多个InputSplitInputSplit是MapReduce对文件进行处理和运算的输入单位只是一个逻辑概念每个InputSplit并没有对文件进行实际切割只是记录了要处理的数据的位置和长度。因为InputSplit是逻辑切分而非物理切分所以还需要通过RecordReaderRR并根据InputSplit中的信息来处理InputSplit中的具体记录加载数据并转换为适合Map任务读取的键值对输入给Map任务。Map任务会根据用户自定义的映射规则输出一系列的key,value作为中间结果。为了让Reduce可以并行处理Map的结果需要对Map的输出进行一定的分区、排序Sort、合并Combine和归并Merge等操作得到key,value-list形式的中间结果再交给对应的Reduce程序进行处理这个过程称为Shuffle。Reduce以一系列key,value-list中间结果作为输入执行用户定义的逻辑输出结果给OutputFormat模块。OutputFormat模块会验证输出目录是否已经存在以及输出结果类型是否符合配置文件中的配置类型如果都满足就输出Reduce的结果到分布式文件系统。 2.3 Shuffle过程详解
2.3.1 Shuffle过程简介
所谓Shuffle是指针对Map输出结果进行分区、排序和合并等处理并交给Reduce的过程。因此Shuffle过程分为Map端的操作和Reduce端的操作。 在Map端的Shuffle过程。Map的输出结果首先被写入缓存当缓存满时就启动溢写操作把缓存中的数据写入磁盘文件并清空缓存。当启动溢写操作时首先需要把缓存中的数据进行分区然后对每个分区的数据进行排序Sort和合并Combine之后再写入磁盘文件。每次溢写操作会生成一个新的磁盘文件随着Map任务的执行磁盘中就会生成多个溢写文件。在Map任务全部结束之前这些溢写文件会被归并Merge成一个大的磁盘文件然后通知相应的Reduce任务来领取属于自己需要处理的数据。 在Reduce端的Shuffle过程。Reduce任务从Map端的不同Map机器领回属于自己需要处理的那部分数据然后对数据进行归并Merge后交给Reduce处理。
2.3.2 Map端得Shuffle过程
输入数据和执行Map任务 Map任务的输入数据一般保存在分布式文件系统如GFS或HDFS的文件块中这些文件块的格式是任意的可以是文档也可以是二进制格式的。Map任务接受key,value作为输入后按一定的映射规则转换成一批key,value进行输出。写入缓存 每个Map任务都会被分配一个缓存Map的输出结果不是立即写入磁盘而是首先写入缓存。在缓存中积累一定数量的Map输出结果以后再一次性批量写入磁盘这样可以大大减少对磁盘I/O的影响。因为磁盘包含机械部件它是通过磁头移动和盘片的转动来寻址定位数据的每次寻址的开销很大如果每个Map输出结果都直接写入磁盘会引入很多次寻址开销而一次性批量写入就只需要一次寻址连续写入大大降低了开销。需要注意的是在写入缓存之前key与value值都会被序列化成字节数组。溢写分区、排序和合并 提供给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任务最终的计算结果一般而言累加、最大值等场景可以使用合并操作。 经过分区、排序以及可能发生的合并操作之后这些缓存中的键值对就可以被写入磁盘并清空缓存。每次溢写操作都会在磁盘中生成一个新的溢写文件写入溢写文件中的所有键值对都是经过分区和排序**的。文件归并 每次溢写操作都会在磁盘中生成一个新的溢写文件随着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任务进行处理具体执行流程如下 “领取”数据 Map端的Shuffle过程结束后所有Map输出结果都保存在Map机器的本地磁盘上Reduce任务需要把这些数据“领取”Fetch回来存放到自己所在机器的本地磁盘上。因此在每个Reduce任务真正开始之前它大部分时间都在从Map端把属于自己处理那些分区的数据“领取”过来。 每个Reduce任务会不断地通过RPCRemote Procedure Call向JobTracker询问Map任务是否已经完成JobTracker监测到一个Map任务完成后就会通知相关的Reduce任务来“领取”数据一旦一个Reduce任务收到JobTracker通知它就会到该Map任务所在机器上把属于自己处理的分区数据领取到本地磁盘中。一般系统中会存在多个Map机器因此Reduce任务会使用多个线程同时从多个Map机器领回数据。归并数据 从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个归并后的大文件。把数据输入Reduce任务 磁盘中经过多轮归并后得到的若干个大文件不会继续归并成一个新的大文件而是直接输入给Reduce任务这样可以减少磁盘读写开销。由此整个Shuffle过程顺利结束。接下来Reduce任务会执行Reduce函数中定义的各种映射输出最终结果并保存到分布式文件系统中。 3. 以WordCount为例理解MapReduce过程 判断WordCount任务是否可以采用MapReduce实现。满足前提条件待处理的数据集可以分解成许多小的数据集而且每一个小数据集都可以完全并行地进行处理核心是并行处理本例中不同单词之间的频数不存在相关性彼此独立可以把不同的单词分发给不同的机器进行并行处理因此可以采用MapReduce来实现词频统计任务。 确定MapReduce程序的设计思路。本例中把文件内容解析成许多个单词然后把所有相同的单词聚集到一起。最后计算出每个单词出现的次数进行输出 确定MapReduce程序的执行过程。把一个大文件切分成许多个分片每个分片输入给不同机器上的Map任务并行执行完成“从文件中解析出所有单词”的任务。Map的输入采用Hadoop默认的key, value输入方式即文件的行号作为key文件的一行作为valueMap的输出以单词作为key1作为value即单词1表示单词出现了1次。 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详细讲解 数据分片 split的map过程 Reduce过程 WordCount的Map过程 WordCount的Reduce过程 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学习资料
