椹川粤语怎么读:分布式基础学习【二】 —— 分布式计算系统(Map/Reduce)
来源:百度文库 编辑:偶看新闻 时间:2024/04/25 19:08:41
分布式基础学习【二】 —— 分布式计算系统(Map/Reduce)
二. 分布式计算(Map/Reduce)
分布式式计算,同样是一个宽泛的概念,在这里,它狭义的指代,按GoogleMap/Reduce框架所设计的分布式框架。在Hadoop中,分布式文件系统,很大程度上,是为各种分布式计算需求所服务的。我们说分布式文件系统就是加了分布式的文件系统,类似的定义推广到分布式计算上,我们可以将其视为增加了分布式支持的计算函数。从计算的角度上看,Map/Reduce框架接受各种格式的键值对文件作为输入,读取计算后,最终生成自定义格式的输出文件。而从分布式的角度上看,分布式计算的输入文件往往规模巨大,且分布在多个机器上,单机计算完全不可支撑且效率低下,因此Map/Reduce框架需要提供一套机制,将此计算扩展到无限规模的机器集群上进行。依照这样的定义,我们对整个Map/Reduce的理解,也可以分别沿着这两个流程去看。。。在Map/Reduce框架中,每一次计算请求,被称为作业。在分布式计算Map/Reduce框架中,为了完成这个作业,它进行两步走的战略,首先是将其拆分成若干个Map任务,分配到不同的机器上去执行,每一个Map任务拿输入文件的一部分作为自己的输入,经过一些计算,生成某种格式的中间文件,这种格式,与最终所需的文件格式完全一致,但是仅仅包含一部分数据。因此,等到所有Map任务完成后,它会进入下一个步骤,用以合并这些中间文件获得最后的输出文件。此时,系统会生成若干个Reduce任务,同样也是分配到不同的机器去执行,它的目标,就是将若干个Map任务生成的中间文件为汇总到最后的输出文件中去。当然,这个汇总不总会像1+ 1 =2那么直接了当,这也就是Reduce任务的价值所在。经过如上步骤,最终,作业完成,所需的目标文件生成。整个算法的关键,就在于增加了一个中间文件生成的流程,大大提高了灵活性,使其分布式扩展性得到了保证。。。I. 术语对照
和分布式文件系统一样,Google、Hadoop和....我,各执一种方式表述统一概念,为了保证其统一性,特有下表。。。文中翻译 Hadoop术语 Google术语 相关解释 作业 Job Job 用户的每一个计算请求,就称为一个作业。 作业服务器 JobTracker Master 用户提交作业的服务器,同时,它还负责各个作业任务的分配,管理所有的任务服务器。 任务服务器 TaskTracker Worker 任劳任怨的工蜂,负责执行具体的任务。 任务 Task Task 每一个作业,都需要拆分开了,交由多个服务器来完成,拆分出来的执行单位,就称为任务。 备份任务 Speculative Task Buckup Task 每一个任务,都有可能执行失败或者缓慢,为了降低为此付出的代价,系统会未雨绸缪的实现在另外的任务服务器上执行同样一个任务,这就是备份任务。
II. 基本架构
与分布式文件系统类似,Map/Reduce的集群,也由三类服务器构成。其中作业服务器,在Hadoop中称为JobTracker,在Google论文中称为Master。前者告诉我们,作业服务器是负责管理运行在此框架下所有作业的,后者告诉我们,它也是为各个作业分配任务的核心。与HDFS的主控服务器类似,它也是作为单点存在的,简化了负责的同步流程。具体的负责执行用户定义操作的,是任务服务器,每一个作业被拆分成很多的任务,包括Map任务和Reduce任务等,任务是具体执行的基本单元,它们都需要分配到合适任务服务器上去执行,任务服务器一边执行一边向作业服务器汇报各个任务的状态,以此来帮助作业服务器了解作业执行的整体情况,分配新的任务等等。。。除了作业的管理者执行者,还需要有一个任务的提交者,这就是客户端。与分布式文件系统一样,客户端也不是一个单独的进程,而是一组API,用户需要自定义好自己需要的内容,经由客户端相关的代码,将作业及其相关内容和配置,提交到作业服务器去,并时刻监控执行的状况。。。同作为Hadoop的实现,与HDFS的通信机制相同,HadoopMap/Reduce也是用了协议接口来进行服务器间的交流。实现者作为RPC服务器,调用者经由RPC的代理进行调用,如此,完成大部分的通信,具体服务器的架构,和其中运行的各个协议状况,参见下图。从图中可以看到,与HDFS相比,相关的协议少了几个,客户端与任务服务器,任务服务器之间,都不再有直接通信关系。这并不意味着客户端就不需要了解具体任务的执行状况,也不意味着,任务服务器之间不需要了解别家任务执行的情形,只不过,由于整个集群各机器的联系比HDFS复杂的多,直接通信过于的难以维系,所以,都统一由作业服务器整理转发。另外,从这幅图可以看到,任务服务器不是一个人在战斗,它会像孙悟空一样招出一群宝宝帮助其具体执行任务。这样做的好处,个人觉得,应该有安全性方面的考虑,毕竟,任务的代码是用户提交的,数据也是用户指定的,这质量自然良莠不齐,万一碰上个搞破坏的,把整个任务服务器进程搞死了,就因小失大了。因此,放在单独的地盘进行,爱咋咋地,也算是权责明确了。。。与分布式文件系统相比,Map/Reduce框架的还有一个特点,就是可定制性强。文件系统中很多的算法,都是很固定和直观的,不会由于所存储的内容不同而有太多的变化。而作为通用的计算框架,需要面对的问题则要复杂很多,在各种不同的问题、不同的输入、不同的需求之间,很难有一种包治百病的药能够一招鲜吃遍天。作为Map/Reduce框架而言,一方面要尽可能的抽取出公共的一些需求,实现出来。更重要的,是需要提供良好的可扩展机制,满足用户自定义各种算法的需求。Hadoop是由Java来实现的,因此通过反射来实现自定义的扩展,显得比较小菜一碟了。在JobConf类中,定义了大量的接口,这基本上是HadoopMap/Reduce框架所有可定制内容的一次集中展示。在JobConf中,有大量set接口接受一个Class extendsxxx>的参数,通常它都有一个默认实现的类,用户如果不满意,则可自定义实现。。。III. 计算流程
如果一切都按部就班的进行,那么整个作业的计算流程,应该是作业的提交 -> Map任务的分配和执行 -> Reduce任务的分配和执行-> 作业的完成。而在每个任务的执行中,又包含输入的准备 -> 算法的执行 ->输出的生成,三个子步骤。沿着这个流程,我们可以很快的整理清晰整个Map/Reduce框架下作业的执行。。。1、作业的提交
一个作业,在提交之前,需要把所有应该配置的东西都配置好,因为一旦提交到了作业服务器上,就陷入了完全自动化的流程,用户除了观望,最多也就能起一个监督作用,惩治一些不好好工作的任务。。。基本上,用户在提交代码阶段,需要做的工作主要是这样的:首先,书写好所有自定的代码,最起码,需要有Map和Reduce的执行代码。在Hadoop中,Map需要派生自Mapper2、Map任务的分配
当一个作业提交到了作业服务器上,作业服务器会生成若干个Map任务,每一个Map任务,负责将一部分的输入转换成格式与最终格式相同的中间文件。通常一个作业的输入都是基于分布式文件系统的文件(当然在单机环境下,文件系统单机的也可以...),因为,它可以很天然的和分布式的计算产生联系。而对于一个Map任务而言,它的输入往往是输入文件的一个数据块,或者是数据块的一部分,但通常,不跨数据块。因为,一旦跨了数据块,就可能涉及到多个服务器,带来了不必要的复杂性。。。当一个作业,从客户端提交到了作业服务器上,作业服务器会生成一个JobInProgress对象,作为与之对应的标识,用于管理。作业被拆分成若干个Map任务后,会预先挂在作业服务器上的任务服务器拓扑树。这是依照分布式文件数据块的位置来划分的,比如一个Map任务需要用某个数据块,这个数据块有三份备份,那么,在这三台服务器上都会挂上此任务,可以视为是一个预分配。。。关于任务管理和分配的大部分的真实功能和逻辑的实现,JobInProgress则依托JobInProgressListener和TaskScheduler的子类。TaskScheduler,顾名思义是用于任务分配的策略类(为了简化描述,用它代指所有TaskScheduler的子类...)。它会掌握好所有作业的任务信息,其assignTasks函数,接受一个TaskTrackerStatus作为参数,依照此任务服务器的状态和现有的任务状况,为其分配新的任务。而为了掌握所有作业相关任务的状况,TaskScheduler会将若干个JobInProgressListener注册到JobTracker中去,当有新的作业到达、移除或更新的时候,JobTracker会告知给所有的JobInProgressListener,以便它们做出相应的处理。。。任务分配是一个重要的环节,所谓任务分配,就是将合适作业的合适任务分配到合适的服务器上。不难看出,里面蕴含了两个步骤,先是选择作业,然后是在此作业中选择任务。和所有分配工作一样,任务分配也是一个复杂的活。不良好的任务分配,可能会导致网络流量增加、某些任务服务器负载过重效率下降,等等。不仅如此,任务分配还是一个无一致模式的问题,不同的业务背景,可能需要不同的算法才能满足需求。因此,在Hadoop中,有很多TaskScheduler的子类,像Facebook,Yahoo,都为其贡献出了自家用的算法。在Hadoop中,默认的任务分配器,是JobQueueTaskScheduler类。它选择作业的基本次序是:MapClean Up Task(Map任务服务器的清理任务,用于清理相关的过期的文件和环境...) -> Map SetupTask(Map任务服务器的安装任务,负责配置好相关的环境...) -> Map Tasks -> Reduce Clean Up Task-> Reduce Setup Task -> ReduceTasks。在这个前提下,具体到Map任务的分配上来。当一个任务服务器工作的游刃有余,期待获得新的任务的时候,JobQueueTaskScheduler会按照各个作业的优先级,从最高优先级的作业开始分配。每分配一个,还会为其留出余量,已被不时之需。举一个例子:系统目前有优先级3、2、1的三个作业,每个作业都有一个可分配的Map任务,一个任务服务器来申请新的任务,它还有能力承载3个任务的执行,JobQueueTaskScheduler会先从优先级3的作业上取一个任务分配给它,然后再留出一个1任务的余量。此时,系统只能在将优先级2作业的任务分配给此服务器,而不能分配优先级1的任务。这样的策略,基本思路就是一切为高优先级的作业服务,优先分配不说,分配了好保留有余力以备不时之需,如此优待,足以让高优先级的作业喜极而泣,让低优先级的作业感慨既生瑜何生亮,甚至是活活饿死。。。确定了从哪个作业提取任务后,具体的分配算法,经过一系列的调用,最后实际是由JobInProgress的findNewMapTask函数完成的。它的算法很简单,就是尽全力为此服务器非配且尽可能好的分配任务,也就是说,只要还有可分配的任务,就一定会分给它,而不考虑后来者。作业服务器会从离它最近的服务器开始,看上面是否还挂着未分配的任务(预分配上的),从近到远,如果所有的任务都分配了,那么看有没有开启多次执行,如果开启,考虑把未完成的任务再分配一次(后面有地方详述...)。。。对于作业服务器来说,把一个任务分配出去了,并不意味着它就彻底解放,可以对此任务可以不管不顾了。因为任务可以在任务服务器上执行失败,可能执行缓慢,这都需要作业服务器帮助它们再来一次。因此在Task中,记录有一个TaskAttemptID,对于任务服务器而言,它们每次跑的,其实都只是一个Attempt而已,Reduce任务只需要采信一个的输出,其他都算白忙乎了。。。3、Map任务的执行
与HDFS类似,任务服务器是通过心跳消息,向作业服务器汇报此时此刻其上各个任务执行的状况,并向作业服务器申请新的任务的。具体实现,是TaskTracker调用InterTrackerProtocol协议的heartbeat方法来做的。这个方法接受一个TaskTrackerStatus对象作为参数,它描述了此时此任务服务器的状态。当其有余力接受新的任务的时候,它还会传入acceptNewTasks为true的参数,表示希望作业服务器委以重任。JobTracker接收到相关的参数后,经过处理,会返回一个HeartbeatResponse对象。这个对象中,定义了一组TaskTrackerAction,用于指导任务服务器进行下一步的工作。系统中已定义的了一堆其TaskTrackerAction的子类,有的对携带的参数进行了扩充,有的只是标明了下ID,具体不详写了,一看便知。。。当TaskTracker收到的TaskTrackerAction中,包含了LaunchTaskAction,它会开始执行所分配的新的任务。在TaskTracker中,有一个TaskTracker.TaskLauncher线程(确切的说是两个,一个等Map任务,一个等Reduce任务),它们在痴痴的守候着新任务的来到。一旦等到了,会最终调用到Task的createRunner方法,构造出一个TaskRunner对象,新建一个线程来执行。对于一个Map任务,它对应的Runner是TaskRunner的子类MapTaskRunner,不过,核心部分都在TaskRunner的实现内。TaskRunner会先将所需的文件全部下载并拆包好,并记录到一个全局缓存中,这是一个全局的目录,可以供所有此作业的所有任务使用。它会用一些软链接,将一些文件名链接到这个缓存中来。然后,根据不同的参数,配置出一个JVM执行的环境,这个环境与JvmEnv类的对象对应。接着,TaskRunner会调用JvmManager的launchJvm方法,提交给JvmManager处理。JvmManager用于管理该TaskTracker上所有运行的Task子进程。在目前的实现中,尝试的是池化的方式。有若干个固定的槽,如果槽没有满,那么就启动新的子进程,否则,就寻找idle的进程,如果是同Job的直接放进去,否则杀死这个进程,用一个新的进程代替。每一个进程都是由JvmRunner来管理的,它也是位于单独线程中的。但是从实现上看,这个机制好像没有部署开,子进程是死循环等待,而不会阻塞在父进程的相关线程上,父线程的变量一直都没有个调整,一旦分配,始终都处在繁忙的状况了。真实的执行载体,是Child,它包含一个main函数,进程执行,会将相关参数传进来,它会拆解这些参数,并且构造出相关的Task实例,调用其run函数进行执行。每一个子进程,可以执行指定个数量的Task,这就是上面所说的池化的配置。但是,这套机制在我看来,并没有运行起来,每个进程其实都没有机会不死而执行新的任务,只是傻傻的等待进程池满,而被一刀毙命。也许是我老眼昏花,没看出其中实现的端倪。。。4、Reduce任务的分配与执行
比之Map任务,Reduce的分配及其简单,基本上是所有Map任务完成了,有空闲的任务服务器,来了就给分配一个Job任务。因为Map任务的结果星罗棋布,且变化多端,真要搞一个全局优化的算法,绝对是得不偿失。而Reduce任务的执行进程的构造和分配流程,与Map基本完全的一致,没有啥可说的了。。。但其实,Reduce任务与Map任务的最大不同,是Map任务的文件都在本地隔着,而Reduce任务需要到处采集。这个流程是作业服务器经由此Reduce任务所处的任务服务器,告诉Reduce任务正在执行的进程,它需要的Map任务执行过的服务器地址,此Reduce任务服务器会于原Map任务服务器联系(当然本地就免了...),通过FTP服务,下载过来。这个隐含的直接数据联系,就是执行Reduce任务与执行Map任务最大的不同了。。。5、作业的完成
当所有Reduce任务都完成了,所需数据都写到了分布式文件系统上,整个作业才正式完成了。此中,涉及到很多的类,很多的文件,很多的服务器,所以说起来很费劲,话说,一图解千语,说了那么多,我还是画两幅图,彻底表达一下吧。。。首先,是一个时序图。它模拟了一个由3个Map任务和1个Reduce任务构成的作业执行流程。我们可以看到,在执行的过程中,只要有人太慢,或者失败,就会增加一次尝试,以此换取最快的执行总时间。一旦所有Map任务完成,Reduce开始运作(其实,不一定要这样的...),对于每一个Map任务来说,只有执行到Reduce任务把它上面的数据下载完成,才算成功,否则,都是失败,需要重新进行尝试。。。而第二副图,不是我画的,就不转载了,参见这里,它描述了整个Map/Reduce的服务器状况图,包括整体流程、所处服务器进程、输入输出等,看清楚这幅图,对Map/Reduce的基本流程应该能完全跑通了。有这几点,可能图中描述的不够清晰需要提及一下,一个是在HDFS中,其实还有日志文件,图中没有标明;另一个是步骤5,其实是由TaskTracker主动去拉取而不是JobTracker推送过来的;还有步骤8和步骤11,创建出来的MapTask和ReduceTask,在Hadoop中都是运行在独立的进程上的。。。IV. Map任务详请
从上面,可以了解到整个Map和Reduce任务的整体流程,而后面要啰嗦的,是具体执行中的细节。Map任务的输入,是分布式文件系统上的,包含键值对信息的文件。为了给每一个Map任务指定输入,我们需要掌握文件格式把它分切成块,并从每一块中分离出键值信息。在HDFS中,输入的文件格式,是由InputFormatV. Reduce任务详情
理论上看,Reduce任务的整个执行流程要比Map任务更为的罗嗦一些,因为,它需要收集输入文件,然后才能进行处理。Reduce任务,主要有这么三个步骤:Copy、Sort、Reduce(参见ReduceTask的run方法)。所谓Copy,就是从执行各个Map任务的服务器那里,收罗到本地来。拷贝的任务,是由ReduceTask.ReduceCopier类来负责,它有一个内嵌类,叫MapOutputCopier,它会在一个单独的线程内,负责某个Map任务服务器上文件的拷贝工作。远程拷贝过来的内容(当然也可以是本地了...),作为MapOutput对象存在,它可以在内存中也可以序列化在磁盘上,这个根据内存使用状况来自动调节。整个拷贝过程是一个动态的过程,也就是说它不是一次给好所有输入信息就不再变化了。它会不停的调用TaskUmbilicalProtocol协议的getMapCompletionEvents方法,向其父TaskTracker询问此作业个Map任务的完成状况(TaskTracker要向JobTracker询问后再转告给它...)。当获取到相关Map任务执行服务器的信息后,都会有一个线程开启,做具体的拷贝工作。同时,还有一个内存Merger线程和一个文件Merger线程在同步工作,它们将新鲜下载过来的文件(可能在内存中,简单的统称为文件...),做着归并排序,以此,节约时间,降低输入文件的数量,为后续的排序工作减负。。。Sort,排序工作,就相当于上述排序工作的一个延续。它会在所有的文件都拷贝完毕后进行,因为虽然同步有做着归并的工作,但可能留着尾巴,没做彻底。经过这一个流程,该彻底的都彻底了,一个崭新的、合并了所有所需Map任务输出文件的新文件,诞生了。而那些千行万苦从其他各个服务器网罗过来的Map任务输出文件,很快的结束了它们的历史使命,被扫地出门一扫而光,全部删除了。。。所谓好戏在后头,Reduce任务的最后一个阶段,正是Reduce本身。它也会准备一个OutputCollector收集输出,与MapTask不同,这个OutputCollector更为简单,仅仅是打开一个RecordWriter,collect一次,write一次。最大的不同在于,这次传入RecordWriter的文件系统,基本都是分布式文件系统,或者说是HDFS。而在输入方面,ReduceTask会从JobConf那里调用一堆getMapOutputKeyClass、getMapOutputValueClass、getOutputKeyComparator等等之类的自定义类,构造出Reducer所需的键类型,和值的迭代类型Iterator(一个键到了这里一般是对应一组值)。具体实现颇为拐弯抹角,建议看一下Merger.MergeQueue,RawKeyValueIterator,ReduceTask.ReduceValuesIterator等等之类的实现。有了输入,有了输出,不断循环调用自定义的Reducer,最终,Reduce阶段完成。。。