hadoop heartbeat分析

2440

　　1. Data node -> name node heart beat
　　datanode 发起heart beat，周期性的向name node发送。heart beat间隔通过 dfs.heartbeat.interval 进行设置。
　　在data node中的主循环 offerService（）发起heart beat： 主要包含以下信息：
　　
　　DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,       //  DatanodeRegistration
                                                       data.getCapacity(),                                //  data node的容量
                                                       data.getDfsUsed(),                                 //  data node已用空间
                                                       data.getRemaining(),                             //  data node剩余空间
                                                       xmitsInProgress,                                    //  传输的block数
                                                       getXceiverCount());                               //   data node xceiver 的数目
　　
　　
　　DatanodeRegistration包括了namenode用来标识个验证data node的全部信息。
　　heart beat 通过rpc，调用 Namenode的同名方法。Namenode 类会调用FSNamesystem 中handleHeartbeat完成heart beat的处理工作。
　　在handleHeartbeat 中，处理过程主要有以下几步：
　　1. namenode 中维护了一个datanode -> block map.的映射关系（对应于数据结构为 StorageID -> DatanodeDescriptor）， 当namenode收到一个heart beat request 后，在该map中查找对应data node是否存在，如果不存在，则返回data node command DatanodeCommand.REGISTER.
　　2. 检测 data node 是否应该被shut down， （即：如果该节点被置为 DECOMMISSIONED， 则其应该被shut down）。如果该data node应该被shut down，则namenode会通过rpc 将DisallowedDatanodeException 丢回给对应data node， data node收到该exception后，自杀。
　　3. 更新 StorageID -> DatanodeDescriptor 表 （datanodeMap） 中DatanodeDescriptor相关信息
　　4. 检查lease recovery
　　5. 检查处于pending 状态的 replication
　　6. 检查非法块
　　7. 检查是否需要升级
　　
　　2. task tracker -> job tracker heart beat
　　同dn -> nn 的heartbeat相似。 tt -> jt的heartbeat是由tt发起的，由tt中offerService() 周期性的调用transmitHeartBeat ()，向jt发送心跳包。heart beat主要包括以下信息：
　　
　　
　　HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status,      // TaskTrackerStatus
　　justStarted,                                   // tt是否为刚启动
　　askForNewTask,                             // tt是否向jt要新的task
　　heartbeatResponseId);                  // 上次心跳response 的id
　　
　　参数TaskTrackerStatus 包含了一个tt详细的信息，包括其上运行的全部task的状态（TaskTrackerStatus），host， tracker name，maxReduceTasks， maxMapTasks，ResourceStatus (tt的资源使用状况).
　　heart beat 通过InterTrackerProtocol 发送到jt， 由jt进行下面的工作。
　　1.检查tt是否为被允许节点（即：是否在slave列表中且不再exclude 列表中）
　　2. jt中维护这trackerID --> last sent HeartBeatResponse的映射关系，取该tt上次heart beat的response.
　　如果tt不是初始连接，如果jt中上次 heart beat resp为空，则表示 jt出现了严重的问题，现在的实现中，会记录一个warning，然后new一个新的resp。
　　3.调用processHeartbeat()， 即根据此次heart beat中所包含的tt 和 task的信息，更新jt中相应的数据结构
　　4.初始化 jt返回的response 对象
　　5.检测将在该 tasktracker上运行的新task
　　6.检测需要被kill的task
　　7.检测需要被kill或者cleanup的job
　　8.更新jt中的 trackerToHeartbeatResponseMap结构
　　9.处理完 hearbeat, 删除该tt上全部 'marked' tasks。