《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类
