《Hadoop、HDFS数据流_解析_第二版 (肖韬_南京大学计算机系)》由会员分享,可在线阅读,更多相关《Hadoop、HDFS数据流_解析_第二版 (肖韬_南京大学计算机系)(46页珍藏版)》请在金锄头文库上搜索。
1、Hadoop Data Flow Hadoop Data Flow 解析解析肖韬 南京大学 计算机系2011.02Job与Nodes的组成一个job由若干task组成: 若干 map tasks 若干 reduce tasks控制job运行的两类nodes:1个 jobtracker:协调和控制系统中运行的所有jobs,以及所有在tasktrackers上运行的tasks若干 tasktrackers:运行task,并向jobtracker发送进度报告(记录了每一个job的运行进度)如果一个task失败了,jobtracker可以将其重新部署到另一个tasktracker上运行JobTrack
2、er与TaskTrackerNamenode与Datanode(1)Namenode与Datanode是HDFS中的概念。被存储在HDFS中的数据以block为单位存储,且每一个block被复制多份存储在不同节点,以提供可靠性保证和高速访问。HDFS采用master-slaves的架构:master管理数据文件的namepace (如metadata,目录结构,文件到blocks的映射,blocks的位置,访问权限等)slaves则管理实际的数据blocksmaster指导client对数据进行访问Namenode与Datanode(2)在GFS中master被称作GFS masterslav
3、es被称作GFS chunkservers在Hadoop中master被称作namenodeslaves则被称作datanodesHDFS架构Hadoop集群的架构Input splits, and recordsMapReduce的输入(input file)被切分为固定大小的input splits,简称splitHadoop为每一个split都创建一个map task,该map task中的map函数会作用于split中的每一个record.一个record就是一个key-value pair这是input fileInput split这是一个Input splitrecord注意:i
4、nput split是对record(即key-value pair)在逻辑上的切分,而HDFS中的block是对输入数据的物理切分。当两者一致时,很高效,但是实际中往往不一致。record可能会跨越block的边界。Split的大小选择Split不该太大(失去parallel性)也不该太小(额外的开销占比过大)与HDFS中的一个Block的大小相同较为合适(Block默认为64BM)移动计 算,而不是移动数据Task在运行时需要数据Job scheduler会在已经有了所需数据的节点上启动对应的task,这样就实现了data localitymap 与 reduce 的输出Map task的
5、输出将被写入磁盘(Linux文件系统),而不是HDFS文件系统。为什么?Map的输出是中间临时结果(intermediate key-value pairs),它们作为reduce tasks的输入一旦job结束,这些中间临时结果即被丢弃,不再需要如果存入HDFS,就需要复制多份副本在网络上传输,浪费!Reduce task的输出会被写入到HDFS文件系统中毕竟,它们的输出是用户最终需要的结果,要妥善保存只有1个reduce task的数据流图每个虚线框表示一个nodepartition当存在多个reducer时,map task会自己的输出进行partition,然后再将其送至reduce
6、task。每一个partition会被发送至某个特别的reduce task一个partition内可以包含多个不同的key,同一个key一定在一个partition内可以生成自定义的partitioner类(继承自)有多个reduce tasks时的数据流图每个虚线框表示一个node可以只有map task,而没有reduce task这种情况下,map task的输出会被写入HDFS深入解析 Shuffle & SortMapReduce保证每一个reducer的输入都是sorted-by-key怎样保证? 通过接口类WritableComparable中的compareTo函数的返回值来
7、进行排序(-1是小于关系,0是等于,1是大于关系)Hadoop进行sort-by-key,并将map的输出传送至reduce的输入的过程,称为ShuffleThe map side (1)每一个map task都有circular memory buffer,map task将自己的输出写入该buffer一个后台线程对buffer中的数据进行partition(根据这些数据将被送至哪一个reduce task)在每一个partition内,后台线程将进行sort-by-key(reduce task收到的数据,都是按照key的顺序排列好的)(如果有combiner,combiner将作用于在上
8、述sort-by-key的输出)The map side (2)当该buffer中写入的内容大小达到某个阈值时,一个后台线程将开始将内容spill到磁盘上(此时map task写buffer的操作仍然继续,buffer如果满了,map task将阻塞直到spill完成)每当buffer的内容达到阈值时(后台线程开始spill时),一个spill file就会被创建。当map task的输出完最后一个record时,就可能存在多个spill files在map task完成之前,这些多个spill files将被归并到单独一个partitioned and sorted的输出文件(output
9、file*)中The map side (3)当指定了一个combiner function、且spill的数量达到3个时,那么在写output file*之前(?),combiner会运行。The reduce side (1)Copy phase多个map tasks可能先后完成,任意一个map task一旦完成,reduce task就开始复制它们的输出(特定的partition),作为自己的输入在reduce task取完自己的输入(即map task的输出)后,这些数据并不是立即删掉,以应对reduce task失败的情况Merge phase当reduce task复制了map t
10、ask的全部输出时,就开始将这些输出归并到少数几个文件(并保持文件内的record原来的有序)The reduce side (2)reducer 怎样知道从哪些map tasktracker那里去取自己所需要的partition(亦即map task的输出)?当map task成功完成后,它会将状态更新通知它所属的tasktracker,该tasktracker进而又会通知其所属的jobtracker。这些通知是通过heartbeat通信机制实现的。这样,对于一个job而言,jobtracker知道map output与tasktracker之间的映射关系。reducer中的一个线程会周期性
11、地向jobtracker询问map output所在的位置,直到该reducer接收了所有的map output.ChainMapperSpeculative execution(1)Case 1:某台节点上的某个task可能失败,Hadoop会将该task进行完全地复制并转移到另一个节点上重新运行Speculative execution (2)Case 2:某个task运行得很慢,同时还有其他可用的计算节点(计算能力可能非常强),Hadoop会将该task的副本复制到新的可用节点上同时运行。最先完成的节点的结果作为该task的最终结果,其它节点的结果丢弃。Shuffle & sort Re
12、duce task开始的时机 (1)只有当全部的mapper发射完了全部的key-value pairs,且所有的intermediate key-value pairs被shuffled和sorted之后,reduce部分才能开始运行。如果违反了这个限定,那么Hadoop就无法保证对应于一个key的所有values都已经被收集了。原文:Note that the reduce computation cannot start until all the mappers have finished emitting key-value pairs and all intermediate ke
13、y-value pairs have been shuffled and sorted, since the execution framework cannot otherwise guarantee that all values associated with the same key have been gathered.Reduce task开始的时机 (2)Hadoop中reducer会一次性地接收到一个key所对应的所有valueCombiner & PartitionerPartitioner作用于map task的输出Combiner:本质上在mapper的输出被发送至red
14、ucer之前进行的一个local reduce. 当指定了combiner之后,它作用于mapper输出的intermediate key-value pairs零次或者若干次(不定),然后再将结果发送至reducer.要求:combiner的输入和输出的类型完全相同,且不论它被调用多少次,都不影响最终的结果(只影响效率)这是在MapReduce概念上的流程图但是Hadoop在实现时,partitioners是先于combiners被执行的combiner的形式combiner类与reducer的类是相似的派生于类重写该类中的reduce函数更好的combinercombiner的局限性不能保
15、证被Hadoop调用即使被调用,也不知道会被调用多少次所以无法预知combiner的实际执行情况更好的方式:使用in-mapper combiningin-mapper combining (以WordCount为例)class MAPPERmethod SETUPH AssociativeArraymethod MAP(docId a, doc d)for all term t d doHt Ht + 1method CLEANUPfor all term t H doEMIT(t, Ht)这样可以保证实现local aggregationPros and consin-mapper com
16、bining 比 combiner 更加高效combiner只能减少在网络上shuffle的intermediate pairs,却不能减少实际被mapper发射的pairsin-mapper combining 要求 mapper 有足够大的内存mapper节点要能够在内存中存储所有input split中的所有key解决之道:计数,每当处理了N个records后就Emit当前结果Mapper中的几个函数void setup(Context context);void map(Key k, Value v, Context context);void cleanup(Context context);被调用的顺序setupmap map cleanupReducer类与之相似map task 与 mapper objectHadoop会为每一个map task创建一个mapper类的实例对象,每一个key-value pair都会调用一次其中的map函数。所以,可以在处理多个records时保存相关的状态信息同理, Hadoop会为每一个reduce task创建一个reducer类