hadoop源码分析与技术

1、大概流基于NIO,ListenerOP_ACCEPT事件,当有客户端连接过来,Accept后,从大概流基于NIO,ListenerOP_ACCEPT事件,当有客户端连接过来,Accept后,从readers中选取一个Reader将客户端C hannel Reader NIO selector,Connection Channel,Reader OP_READ 事件.客户端建立连接后,ConnnectionHeader 包含协议名,用户组信息,验证方法,Connection 4 位的长度代表这次请求的数据的长度,够长度,为调用id param,新建一个Call 对象(关联Connection)放

2、入call 队列中,handlers 中的Handler 会将Call call Quene 中取走,Call 中的 param 实际为 Invocation 对象,包含调用方法名,参数名,参数类型,由这些信息使用 Java 反射 A pond方法,Call加入到Connection的responseQuene中,如果responseQuene1做一次NIO写操作,能能够将数据写完,responder OP_WRITE,下一次继续写,channel responder了直接加入队列,responder 线程后续处理NOTE:客户端关闭流后出发一次读操作,返回为-1,ServerCurCall与

3、获取客户HandlerCall后,Server的curCall(ThreadLocalchannel responder了直接加入队列,responder 线程后续处理NOTE:客户端关闭流后出发一次读操作,返回为-1,ServerCurCall与获取客户HandlerCall后,Server的curCall(ThreadLocal类型)Call,Instance方法实际是在H andler 线程中,Instance Server IP空闲连如果一定时间 到数据并且当前连接现有Call数目为0,则视为空闲连接, nt.kill.max个连接,如果出现OutOfMemoryErrorListen

4、er强全部空闲连,长时间未发送到客户端的响responderCall 如果长时间没有发送到客户端,掉涉及参callQuene队列大小,随集群增大而增大, readersTCP NAGLES handler个数,由构造参数指定,DN,TTsocketsocket设置socket.bind(address, BACKLOG) 总连接数超过多少后,IPC/RPC 模式,调用RPC.getProxy对象,当调用方法的时候实际调用的是Invoker, 将协议,调用的方法名,参数,参数类型封装成Invocation 对象经过cIPC/RPC 模式,调用RPC.getProxy对象,当调用方法的时候实际调用

5、的是Invoker, 将协议,调用的方法名,参数,参数类型封装成Invocation 对象经过c返回流,id,Connection线程负责时候如果超时Server 返回值,的过程中,调用nt的线程会wait 直到Connection 获取到返值,如果没有出现IOException 版本控1,RPC 高低版本头高低版本协议方法变动数据序列化反序列Name 数据序列化反序列Name 协 DatanodeProtocol:DNNN通讯协议NamenodeProtocol: SN NN 通讯协议,NN fsimage amesystem数据结Gset Set get 方法,BlockInfo,Bloc

6、k元素,问题,amesystem数据结Gset Set get 方法,BlockInfo,Block元素,问题,HashMap ReHash BlocksMapLightWeightGSet 时候,会根据JVM内存将数组的大小初始为最大能占用的内存数(4194304 -算法, LightWeightGSet 在BlockInfo 数量比较小的时候get 性近数组BlockInfo继承Block,hashCode 和equals 方法,在Block 中equals 方法只要求传入的对象是Block lockId相等,就认为两个对象相等,故BlockInfo时候分配的在数组中的Index和Get时

7、候由BlockhashCodeBlockMaps Block Block Block 3 long blockId(随机生成)numBytes(块大小BlockInfoBlock2 个属性,LightWeightGSet LinkedElement triplets:3Xreplication的数组,replication 个组,3 个元素,DatanodeDescriptor,Block,DNblockInfo DatanodeDescriptorblockList指向一个BlockInfo,因为每个BlockInfo中的tripletsN上的上一个,BlockInfo,BlockInfo

8、是一个双向链表着对应的新建文件打开输入流后,写入,namenodeBlockInfo,Block DN 后,DN NN,NN triplets (例子前提假设:新建的集群没有文件,DN1 上,DN1,bkl_* 实际为随机数namenodeBlockInfo Gset 中,blk_1,0-64M,DN1namenodeBlockInfo Gset 中,blk_1,0-64M,DN1 blockList DN1 Block blk_1 ,NN blocksMap Block blk_1 BlockInfo blockInfo1 的可用组(=null)DN1(DatanodeDescriptor1

9、),DN1blockListblockListblockInfo1 NN blockInfo2,DN1 NNBlock blk_2,NN blockInfo2 1,triplets的可用组(=null)DN1(DatanodeDescriptor1) 2,blockList blockInfo1,2,blockInfo1DN1blockInfo2 4,DN1 blockList blockInfo1NameNode/DataNode数结NameNode/DataNode数结snapshothadoop fs,如果直接调用程序接口,没这个功能hadoop fs,如果直接调用程序接口,没这个功能与

10、普通文件致eger.MAX_VALUE ,2可做配置INodeComparable 接口,用来排序,查找INodeComparable 接口,用来排序,查找时候按照大小顺序,查找每层二分查找. Block 3 long blockId(随机生成)numBytes(块大小BlockInfoBlock2 个属性,LightWeightGSet LinkedElement triplets:3Xreplication的数组,replication 个组,3 个元素,DatanodeDescriptor,Block,DNblockInfo DatanodeDescriptorblockList指向一个

11、BlockInfo,因为每个BlockInfo中的tripletsN上的上一个,BlockInfo,BlockInfo 是一个双向链表着对应的节点选HDFSsplacementpolicyistoputonereplicaononehe localrack,another ona nodein a different (remote) andthe节点选HDFSsplacementpolicyistoputonereplicaononehe localrack,another ona nodein a different (remote) andthelastonadifferenthesam

12、eremoteNN DN. 同时满足1,DN 状态正常,退役中,3,DN负载状态,通过有多少活动连接计算,DataXceiver 的个数(尽量均匀,并满足上述第二节点写文件流k(数据和校验信息(0 心跳包1,RPCNN请求建立一个文件,NN做一系列的检查,通过后将文件信息 到 EditLog 中,flush 到硬盘. 2,RPCNNblock,NNblockIdgenerationSt ,n(副本)DN 3,DN,PipeLine,DN,发出传输头信息,InputStreamDN端口,传统模式, block数据流(不同于RPC处理小数据量,元数据的通信),接,DataXceiver线程处理,

13、DataXceiver根据,报文头中的操作指令(read block ,write block)具体的操作,write block,3 1,blobksBeingWrittenblock文件,如果出错会抛出异常,关闭流,3 2,取第一个剩余节点,作为下一个节点,向下一个节点传送重新封装(改变报文中的剩余节点)的报文头,如果出错3,InputStream,如果出错,DN,如果没出错将4,DataStreamerdataQuene取出Packet传输,ackQuene中,ResponsePro ket ackQuene 中移除.sor 处理反馈信息,接收到一个5,DNDataXceiver接受到P

14、acket后会根据Packet中的检验区和数据区校验数据,重要每次写的时候是根据offsetI nBlockmeta文件的位置写入(处理情况, dn2 挂掉后,c nt dn3 packet_X 的反馈,恢复 PipeLine后继续向dn3packet_X),flush到硬盘设置可见长度visibleLength,Packet发送到下一个节点,然后将P acketackQuenePacketResponder处理, PacketResponder接受下一个节点的Packet的反馈,个节点反馈,PacketBlock的最后一个,meta 文件,block flush 到硬盘,关闭.NameNod

15、e ock 信息(并没有),NameNode block-blockInfo 对应.block,无论你写了多大,sync,close前的数据是保险的64M一个块,200M,sync, close NN 重启,正在写的块的处理,block,无论你写了多大,sync,close前的数据是保险的64M一个块,200M,sync, close NN 重启,正在写的块的处理,无blockInfo对应,建立PipeLine 的时候,NN 分配DN 时到InodeFileUnderConstruction 中反馈超时状态InputStream 时候会有一个超时参数,X 3 * 租约在此过程中,DFSC nt

16、 软超时时间限制内主节点负责恢复块过程DN,block的写线程,block 的元信息,主要是大小.(然后比较其大小,ck 的长度.)NameNode ,block ,DN,DN操作:修改meta 文件名,meta 文件.NameNode block 为保持一致性,DN上挂掉之后重新加入集群,block 的例如:DN block 移动到current ,DFSC nt 由于网络原因(DN 断网了)没有收到DN 的反馈,认为DN 挂掉,恢复块,.PipeLine时候出通知NN抛弃建立的Block,将重新请求NN block,DN传递给NN,NN 除传输数据时候DN出 取最小的节点(compare

17、by host:port)作为主节点,开始恢复块的过程DN,DNPipeLine,recoveryFlag1,如果文件已经移动到了current目录,涉及到升级,current, previous目录下为硬连接.block,取最小的节点(compare by host:port)作为主节点,开始恢复块的过程DN,DNPipeLine,recoveryFlag1,如果文件已经移动到了current目录,涉及到升级,current, previous目录下为硬连接.block,detached目录下,然后重命名为current目录下的文件名.返回供写.previous最小的块与重新发送的逻辑Pac

18、ket同一时间,最小的副本,会大于已经发送的长度(dataQuene中已经发完,ackQuene中移除了),而写的时候是按在nt 出DFSOutputStream amesystem 会移除租约文件关闭日志.可能是在写的过程中出错ystem就没有关闭.NameNode租约失效之后, Lease Monitor会检查其中过了硬租约(软超时是要有nt去createappend 一个文件时才会去判断)期限的文件,开始做收尾工作1:block 写入移除租约, InodeFileUnderConstruction关闭文件日志,检查下副本数2: block NN InodeFileUnderConstru

19、ction block s DNA_RECOVERBLOCK 指令发给主节点,主节点做具体恢复操作NameNode出如果在写入块的过程中,NameNode出错,没有影响.当写下一个块的时候,DFSC nt会因为获取不到分配的节点而因为没有关闭文件,NameNode 启动的时候会将这个文件作为正在创建的文件,恢复其租约,失效的时候发起恢复syncblock 信息不会保存到日志文件中,nn 挂掉重启之后当前正在写的块丢失Append 文OPEN INodeFile )(3 个副本,DN 不可用,后来又加入集群中HDFS append recoverLease 接口，能够在出现异常的时候，主动恢复文

20、件，而不等到租NN 发起。Append时候第一次建立PipeLine时候出恢复的时候保留长度,读文APIHDFS append recoverLease 接口，能够在出现异常的时候，主动恢复文件，而不等到租NN 发起。Append时候第一次建立PipeLine时候出恢复的时候保留长度,读文API 介绍read(byte(block)lenbuf offset,的字节数NN block 的信息这里并没有按照树形结构来算. NNblockReaderlocations进行排序. Reader block 在同一机器上,Reader block 在同一机架上.Reader 不在集群中随机首先从NN获

21、取副本信息,如果文件还正在创建中,reader联系返回的第一个dnblock的长度,与nn返回根据当前位置,block,block按(优先)DN,建立连接:short:2)+(type:byte:1)+(,IOException的话,.A 68M,hecker负责校验数据,(seek 512+256 字节处1024 个字节,512 字节-2048 字节,按校验单元来传送的 seek 512+256 字节处1024 个字节,512 字节-2048 字节,按校验单元来传送的 65536,否则(4096)来限制小于的一个 packet 发完大于的分 packet 发送所以传送的可能是第 512 字节

22、到(4096) 个字节(NIO 效率seek 的优化 销大,seekTCP 缓冲大小(128k 取的常见值),这段数据,然后忽略BlockReaderLocal在本机的副本,203 老方法:通过流read1 一定量的数据,然后根据返回的字节数,作block的副本,dndn,选择策略和写入时候一致,加入将dn dn 执行.NN 中,NN 安全模命名空间只读,不进行副启动时候自动进入安全模式手动进入,bin/hadoop dfsadminsafemode 此时如果有文件正在写会出错(和NN通信时候)保存namespace 数据重新平衡策5dfs.balance.bandwidthPerSec 如图

23、中界限,优先选择负载小的,优先选择同一机架的对于多机架,可能会影响,块分布策略扩容和退如图中界限,优先选择负载小的,优先选择同一机架的对于多机架,可能会影响,块分布策略扩容和退退服: dfs.hosts.exclude DN 取出如果其中所有的块都达到了副本数,退役,否则等待当满足退役条件后,dn dn 关闭自己抽象类,1,setReadyToFlushflush,flush的同时,可以继续写入流. 2,flush 后保证数据已经持久化.器JournalProtocol协议,当活动的NameNode 机架感while$#-ne0 case$1机架感while$#-ne0 case$140|41

24、|42)echo43|44|45)echo*)echo/unknownrack#shdns.sh4245命令行运JVM启动参数,lib,conf classpath,JVM RunJar阶段1,MainClassjar Manifest,MainClass 退出时候删除3,Jar , tmpDir),钩子 4,将目录hadoop-unjar38923742,目录hadoop-unjar38923742/class, 目录hadoop-unjar38923742/lib中的每个文件URLClassLoader 参数,classLoader.5,ClassLoader6,5mainClass 启动

25、做准备,mainClass,args bin/hadoop jar -libjars testlib.jar -archives test.tgz -files file.txt inputjar bin/hadoop jar -libjars testlib.jar -archives test.tgz -files file.txt inputjar 均将KVConf实例中(传值,例如将-file指定的文件设定到Conf中,submit中获取,Job.submithdfs写入以下信息目录下文件到-(Split信息)writeSplits -( Split 信息元数据,版本,个数,索引)由w

26、riteSplits -conf-参数-files 逗号分割,交给(DistributedCache 管理-参数-archives 逗号分割, 交给(DistributedCache 管理-参数-libjars 逗号分割, 交给(DistributedCache 管理job.splitmetainfoSplit的在那几个机器上有副本,JTjob.splitSplit,不保存位置信息,TT 不需要(hdfs 决定这个文件用,TaskTaskTT启动时候,mapLauncher(MapTask),reduceLauncher(ReduceTask), TT JTHeartbeatResponse,

27、TaskTrackerAction,5启动任务,LaunchTaskActionTaskConfTaskLauncher LauncherreduceLauncher 中的队列中 启动任务,LaunchTaskActionTaskConfTaskLauncher LauncherreduceLauncher 中的队列中 um 设定,slot 意思为: 槽,TaskTractor 的资源抽象化,um, taskslot,Task也可以通过参数来控制(CapacityTaskScheduler设置,未开放给User?)nJob相关的任务,tasksToCleanup提交任务(1, speculat

28、ive execution 2,need commit REINIT_TRACKER JVM(TTJVM,错误处理,GC)Child.class,main Map 过MapTaskhdfsjob.split JT splitIndex InputSplit,Map 过MapTaskhdfsjob.split JT splitIndex InputSplit,jobConf MapInputFormat,InputFormatRecordReader inputSplit,original_key, 交给Mapper.map方法处理生成gen_key,gen_value,partitionerp

29、artition,成对的(gen_key,gen_value,会先放入一个缓冲区,如图,这个缓冲区分为3级索引(排序缓冲区慢之后,SplitThread 会标记当前前后界,对界内数据进行排序(partition kv),partition 段的位置,部分存到内存部分存到磁盘,在这个过程中,map 仍然继续进行,满之后,mapwait,SplitThread 完毕完毕,随之的SplitThread也结束后,间文件为split.1.outsplit.n.out索引部分存在内存里面, 然后通过合并排序将分段的文件(split.x.n)合并排序成一个文件 partition 信息.( 然后通过合并排序将分段的文件(split.x.n)合并排序成一个文件 partition 信息.(MergeQueue)这样在Reduce过程中,通过http请求TTpartition段(参数reduce),TTfile.outpa