Map Reduce

　　Map Reduce应用包含特定流程:首先编写map和reduce函数.然后写一个驱动程序来运行作业.

　　Hadoop中的组建是按照Hadoop自己的API配置来进行配置的.一个Configuration类的实例,包括配置属性及其值的集合.每一个属性都是String类型,值类型可能是Java的基本类型或String,Class,java.io.File和String Collection.类型的配置信息并不存储在xml中.在读入属性时,属性可以自动识别成一个给定的类,并且get()方法允许一个默认的值,当xml中没有定义时,系统默认使用默认值.标记为final的属性不能被后来定义所覆盖

　　GenericOptionParser是一个解释普通Hadoop命令行选项的类,可以根据应用需要在Configuration对象中设置,通常不用GenericOptionParser,因为实现Tool接口,用ToolRunner更方便

　　作业,任务和任务尝试ID(task attemp id)

　　　　作业ID的格式由job tracker执行的时间和一个有job tracker维护的单独标识作业job tracker实例递增的计数器组成.作业编号由1开始

　　　　　　job_200904110811_0002    —— job tracker运行的第二个作业,开始于2009年4月11日8点11分

　　　　任务属于某个作业,它们的ID即把作业的前缀换成业务的前缀,再加一个后缀来标识业务中的任务(任务ID以0为起始)

　　　　　　task_200904110811_0002_m_0000003  —— ID为job_200904110811_0002作业中第4个任务

　　　　任务尝试在作业执行期间根须需要进行分配,若作业在job tracker重启之后重启,并恢复运行中的作业,那么最后的尝试ID从1000开始计数

　　　　　　attemp_200904110811_0002_m_000003_0

　　

　　使用远程调试器

　　　　当一个任务失败并且没有足够记录来诊断错误时,可以设置运行作业的属性来指导Hadoop保留作业运行期间产生的所有中间值这些数据可以单独在调试器上重新运行那么出错的任务.首先将keep.failed.task.files中的配置属性设置为true.以便在任务失败时,taks tracker能够保留足够多的信息让任务在相同的输入数据上重新运行.然后再次运行作业并使用Web用户界面看任务时在哪个几点失败的(看尝试任务).接着运行IsolationRunner.用前面保留的文件作为输入登录到任务失败的节点.寻找任务尝试目录.由mappred.local.dir属性来设置.这个目录包含多个文件和子目录.包括job.xml,包含任务尝试期间生效的所有作业配置属性.这些属性将被IsolationRunner用来创建一个JobConf实例.对于map任务,这个目录还包含了一个含有输入划分文件序列话表示的文件.对于reduce任务,则由一个map输出备份存储在output目录中.work目录时任务尝试的工作目录,将其改为以这个目录运行IsolationRunner之后,设置断点,连接远程调试.

　　客户端:提交Map Reduce作业

　　job tracker: 协调作业的运行,主类时JobTracker

　　task tracker: 运行作业划分后的任务,主类时TaskTracker

　　HDFS: 在其他实体间共享作业的文件　　

　　JobClient的submitJob()方法所实现的作业提交过程如下:

　　　　1.向job tracker请求一个新的作业ID,即JobTracker.getNewJobID()

　　　　2.检查作业的输出说明,若没有指定输出目录或已存在,作业将不会被提交,并将错误返回给MapReduce程序

　　　　3.计算作业的输入划分,若划分无计算,作业将不会被提交,并将错误返回给Map Reduce程序

　　　　4.将运行作业所需要的资源(作业JAR文件,配置文件,计算所得的输出划分)复制到一个以作业ID号为命名的目录中.jobtracker的文件系统

　　　　5.告诉jobtracker作业准备执行.JobTracker.submitJob()

　　作业的初始化

　　　　JobTracker接收到其submitJob()方法的调用后,会把此调用放入一个内部分队列中,交由作业调度器进行调度,并将其进行初始化.初始化包括创建一个代表该正在运行的作业的对象,它封装任务和记录信息,以便跟踪任务的状态和进程.要创建运行任务列表.作业调度器先从共享文件系统中获取JobClient已计算好的输入划分信息.然后为每个划分创建一个map任务.创建的reduce任务的数量由JobConf的mappred.reduce.task属性决定.然后调度器便创建这么多reduce任务来运行,任务在此指定ID号.

　　任务分配

　　　　TaskTracker执行一个简单的循环,定期发送心跳方法调用JobTracker.心跳告诉JobTracker, TaskTracker是否还存活.同时也充当两者间之间的消息通道.作为心跳方法调用过的一部分,TaskTracker会指明他是否已经准备运行新的任务.若是,jobtracker会为它分配一个任务,并使用心跳方法的返回值与tasktracker进行通信

　　　　在JobTracker为TaskTracker选择任务前,jobtracker必须先选定任务所在的作业.默认情况下,jobtracker维护一个作业优先级别列表.选择作业后,jobtracker就可以为改作业选定一个任务.针对map任务和reduce任务,taskTracker又固定数量的槽,默认调度器在处理reduce任务槽之前会填满空闲的map任务槽.因此tasktracker至少有一个空闲任务槽.jobtracker会为它选择一个map任务,否则选择一个reduce任务.要选择一个reduce任务.jobtracker只是简单地从向未运行到reduce任务列表中选取下一个来执行,并没有考虑数据的本地化.然而相对于一个map任务,它考虑的是tasktracker的网络位置和选取一个距离器输入划分文件最近的tasktracker.在理解情况下,任务是data-local的,与分割文件所在节点运行在相同的节点上.同样,任务可能也是task-local的,和分隔文件在同一个机架上,但不在同一个节点.

　　任务的执行　　

　　　　首先,它本地化作业的JAR文件,将它从共享文件系统复制到tasktracker所在的文件系统.同时,将应用程序所需要的全部文件从分布式缓存复制到本地磁盘,然后为任务新建一个本地工作目录,并把JAR文件中的内容解压到这个文件夹下.新建一个TaskRunner实例来运行任务.TaskRunner启动一个新的java jvm来运行每个任务.使得用户定义的map和reduce函数的任何的任何缺陷都不会影响tasktracker.但在不同任务间重用JVM还是可能的.子进程通过umbilical接口与文件进程进行通信,它每隔几秒便告知文件进程它的进度,直到任务完成.

　　流和管道都运行特殊的map和reduce任务,目的是运行用户提供的可执行程序,并与之通信.应用流时,流任务使用标准输入和输出流.与进程进行通信.另一方面,管道任务则监听套接字,发送环境中的一个端口给C++进程.Java进程都会把输入键/值对传给外部进程,后者通过用户定义的map或reduce函数来执行它并把输出的键/值对传回给Java进程.

　　进程和态度的更新

　　　　Map Reduce的作业和它的每个任务都有一个状态,包括作业或任务的状态(运行,成功,完成,失败),map和reduce的进度,作业计数器的值,状态消息或描述.任务正在运行时,对任务进度保持追踪.对于map任务,是已处理完成输入的百分比,对于reduce任务,仍会估计reduce输入已处理的百分比.整个过程与shuffle的三个结对相对应.

　　　　构成Hadoop进行的所有操作如下:读如一条输入记录(在mapper或reducer中);写入一条输出记录(在mapper或reducer中);在一个报告中设置状态描述(在Reporter()的setStatus()方法);增加一个计数(使用Reporte的incrCounter()方法);调用Reporter的progress()方法

　　　　若任务报告了进度,便会设置一个标志以表明状态变化将被发送到tasktracker.在另一个线程中,每隔三秒检查次标志一次.若已设置,则告知task tracker当前任务状态.同时,task tracker每隔五秒钟发送心跳到job tracker,并且在此调用中,所有由task tracker运行的任务.它们的状态都会发送至job tracker.计数器的发送间隔通常大于5秒.job tracker将这些更新合并起来,产生一个全局视图,表明正在运行的所有作业及其所含任务的状态.JobClient通过每秒查看jobtracker来接收最新的状态.客户端也可以使用JobClient的getJob()方法来得到一个RunningJob的实例,包含所有状态信息.

　　作业的完成

　　　　job tracker收到作业最后一个任务已经完成的通知后,便将作业的状态设置成为“成功”.然后在JobClient查询状态时,它将得知任务已经成功完成,便显示一条消息通知用户,从runJob()返回.

　　失败

　　　　任务失败

　　　　　　子任务失败最常见的情况是map或reduce任务中的用户因为代码抛出的运行时异常.若发生这种情况,子任务jvm进程会在退出之前向其task tracker父进程发送错误报告.错误报告最后被记入用户日志.task tracker会将此任务尝试(task attempt)标记为failed,释放一个槽以便运行另一任务

　　　　　　对于流任务,若流程以非零退出代码退出运行,会被标记为failed.这是由stream.non.zero.exit.is.faillure=true决定的

　　　　　　对于jvm突然退出,可能有JVM错误,由MapReduce用户代码某些特殊原因而造成JVM退出.在这种情况下,task tracker会注意到进程已经退出,并将此尝试标记为failed.

　　　　　　对于任务的挂起,task tracker注意到已经有一段时间没有新的任务,会将任务标记为failed.在此之后,子JVM进程将被自动杀死.任务失败的超时时间间隔通常为10分钟

　　　　　　job tracker被通知一个任务失败时,它将重新调度该任务的执行.job tracker会尝试避免重新调度之前失败过的task tracker上的任务.此外,若一个任务的失败次数超过4次,他就不会再被重试.针对map任务,可以由mapred.map.max.attempts属性设置.对于reduce任务,则由mapred.reduce.max.attempts属性设置.同时还可以通过mapred.max.map.failure.percent和mapred.max.reduce.failure.percent属性来设置在不触发任务失败的情况下,允许任务失败的最多次数.

　　　　task tracker失败

　　　　　　若task tracker由于崩溃或运行过于缓慢而失败,它将停止向job tracker发送心跳.job tracker会在接收心跳超时后将它从等待任务调度的tasktracker池中移除.同时会安排此tasktracker上已经运行并成功完成的map任务返回.若tasktracker失败的次数远高于集群平均任务失败的次数,他就会被放入黑名单.被放入黑名单中的task tracker可以通过重启从job tracker的黑名单中被移除

　　　　job tracker失败　

　　　　　　job tracker失败是一个单点故障

　　　　

　　作业的调度

　　　　可以同时设置mapred.job.priority属性或JobClient的setJobPriority()的方法来设置优先级,但在FIFO调度中,优先级并不支持抢占

　　　　Fair Scheduler的目标是让每个用户的平均的共享集群.若只有一个作业在运行,它得到整个集群的所有资源.作业被放入池中.默认情况下,每个用户都有自己的池.提交作业超过其他用户的用户,不会因此而比其他用户获得超过平均值的集群资源.可以用map和reduce的槽数来定义用户池的最小容量.也可以设置每个池的权重.Fair Scheduler支持抢占.若一个池在特定的一段时间內未能得到公平的资源分配.调度器就会终止运行池中的得到过多资源的任务,以便把槽给资源不足的池

　　shuffle和排序

　　　　MapReduce保证每个reducer的输入都已按键排序.系统执行排序的过程——map输出传到reducer作为后者的输入,即成为shuffle(混洗或洗牌).

　　　　map端

　　　　　　map函数利用缓冲的方式写到内存,并处于效率的原因预先进行排序.每个map任务都有一个环形内存缓冲区,任务会把输出写到此.默认情况下,缓冲区的大小为100mb,此值可以通过io.sort.mb属性来修改.当缓冲区内容达到制定大小时(io.sort.spill.percent,默认为80%),一个后台线程便开始把内容溢出写到磁盘中.在线程工作的同时,map输出继续被写到缓冲区,但若此时缓冲区被填满,map会阻塞直到溢写过程结束

　　　　　　溢写将按轮询方式写到mapred.local.dir属性制定的目录,在一个作业相关子目录中.在写到磁盘之前,线程首先根据数据最终被传送到reducer,将数据划分成相应的分区.在每个分区中,后台线程按键进行內排序(in-memory sort).此时,若有一个combiner,它将基于排序后输出运行.

　　　　　　一旦内存缓冲区达到溢写阈值,就会新建一个溢写文件.因此在map任务写入其最后一个输出记录后,会有若干溢写文件.在任务完成之前,溢写文件被合并成一个已分区且已排序的输出文件.配置属性io.sort.factor控制着一次最多能合并多少流,默认时10.若已制定combiner,并且溢写次数至少为3(min.num.spills..for.combine属性的值)时,combiner就在输出文件被写之前执行.运行combiner的意义在于使map输出更紧凑,从而只有较少数据被写到本地磁盘然后传给reducer.

　　　　　　map输出被写到磁盘时默认情况使不压缩的.可以将mapred.compress.map.outout设置为true,就可以压缩输出.reduer通过HTTP得到输出文件的分区.用户服务于文件分区的工作线程,其次数量由任务的tracker.http.threads属性来控制.此设置针对的使每个task tracker,而不是针对每个map任务槽.默认是40.

　　　　reduce端　　

　　　　　　reduce任务需要为其特定分区文件从集群上若干个map任务的map输出.map任务可以在不同时间完成,因此只要一个任务结束,reduce任务就开始复制其输出.reduce任务由少量复制线程,因此能够并行地获取map输出.默认是5个线程(mapred.reduce.parallel.copied).

　　　　　　map任务完成后,会通知其父tasktracker状态已更新,然后tasktracker进而通知jobtracker.对于制定的作业,jobtracker直到map输出和tasktracker之间的映射关系.reducer中的一个线程定期向job tracker获取map输出位置.直到得到所有输出位置.tasktracker并没有在第一个reducer检索之后就立即从磁盘上删除map输出,因为reducer可能失败.反之,它们会等待,直到被jobtracker告知可以删除.

　　　　　　若map的输出相当小,则会被复制到reduce tasktracker的内存中,否则被复制到磁盘中.内存缓冲区达到阈值大小或达到map输出阈值时,会被合并,进而被溢写到磁盘中.随着磁盘上积累的副本愈来愈多,后台线程会将它们合并为一个更大的,排好序的文件.任何压缩的map输出,都必须在内存中被解压缩,以便合并

　　　　　　所有map输出被复制期间,reduce任务进入排序阶段,这个极端将合并map输出,维持其按顺序排序.最后阶段,即reduce阶段,合并直接把数据输入reduce函数.最后的合并即可来自内存,也可来自磁盘.在reduce阶段,对已排序输出中的每个键依次调用reduce函数.此阶段的输出直接写到输出文件系统,一般为HDFS.若采用HDFS,由于tasktracker节点也运行数据节点,所以第一个块副本会被写到本地磁盘.

　　　　优化的一般原则是为shuffle制定尽量多的内存空间.然而还要确保map和reduce函数能够由足够的内存来运行.为map和reduce任务运行的JVM制定的内存代销由mapred.child.java.opts属性来设置.应让任务节点上这个内存大小尽量大.在map这一端,可以通过避免多次磁盘溢写来获取最佳性能.若可以,应增加io.sort.mb的值.MapReduce计数器将计算在作业运行过程中溢写到磁盘中的记录总数.在reduce这一端,当中间值能够全部存放在内存中,就能获得最佳性能.

　　任务的执行

　　　　MapReduce模型将作业分割成任务,然后并行运行任务,使作业的整体执行时间少于顺序执行的时间.这使得作业执行时间对运行缓慢的任务很敏感.当一个任务比预期要慢时,Hadoop会进行检测,启动另一个相同的任务作为备份.这就是所谓的任务的推测式执行.推测式执行的任务只有在一个作业所有任务都启动之后才启动,并且推测式执行任务只针对已运行一段时间且比作业中其他任务平均进度慢的任务.一个任务成功完成后,任何正在运行的副本任务都会被终止.

　　任务JVM重用

　　　　启用JVM重用后,任务并没有同时运行在一个JVM中.JVM顺序运行任务.尽管task tracker可以一次运行多个任务,但都运行在独立的JVM中.由mapred.job.reuse.jvm.num.tasks指定作业每个JVM运行的task任务的最大数量.默认为1.共享JVM的另一个用处使作业各个任务间的状态共享,将相关数据存储到一个静态的字段后,任务就可以较快速访问共享数据

　　

　　跳过坏记录

　　　　处理错误记录的最佳位置使mapper和reducer代码.可以检测出错误记录并忽略它,或通过抛出一个异常来取消这个作业.还可以使用计数器来计算作业中错误记录的总数,从而查看问题所影响的范围.

　　　　极少数情况下bug存在于第三方的库中,且无法在mapper或reducer中修改它.此时可以用skipping模式选项来自动跳过错误记录.skipping模式启用后,任务将正在处理的记录报告个task tracker.任务失败时,task tracker重新运行该任务,跳过异常任务失败的记录.由于增加网络流量和错误记录的维护,只有在任务失败两次后才会启用skipping模式.

　　　　默认情况下,skpping模式使关闭的,可以使用SkipBadRedcord类单独为map和reduce任务启动它.每次进行任务尝试,skipping模式都只能检测出一个错误记录,因此这种机制仅适用于检测个别错误记录.Hadoop检测出来的记录以序列文件的形式保存在作业输出目录中的_logs/skip子目录下

　　任务执行环境

　　　　确保同一个任务的多个实例不会尝试向同一个文件进行读写操作.需要避免两个问题.第一个问题是若任务失败并被重试,那么在第二个任务运行时原来的输出部分依旧是存在的,所以应先删除第一个任务的旧文件.第二个是在启用推测模式执行的情况下,同一任务的两个实例会同时向同一个文件进行写操作.Hadoop通过将输出写到任务尝试指定的临时文件,解决了任务的常规输出问题.这个目录是{mapred.out put.dir}/_temporart/${mapred.task.id}.若任务执行成功,目录的内容就被复制到此作业的输出目录(${mapred.out put.dir}).因此,若一个任务失败并重试,第一个任务尝试的部分输出就会被清除.一个任务和该任务的推测实例位于不同的工作目录,并且只有先完成的任务才会把其他工作目录中的内容传到输出目录,其他的都被丢弃

　　　　任务完成时提交输出的方法由一个OutputCommitter来实现,它与OutputFormate相关联.FileOutputFormat的OutputCommitter是一个FileOutputCommitter,后者实现了前面描述的提交规则.OutputFormat的getOutputCommitter()方法也会被覆盖以返回一个自定义的OutputCommitter,以免用不同的方式来实现提交过程

　　　　一个任务可以通过检索其配置文件中mapred.work.output.dir属性的值来找到它的工作目录.或使用FileOutputFormat的getWorkOutputPath()静态方法以得到代表工作目录的Path对象.