首页  流程图  详情

03- hadoop HDFS 中Namenode、DataNode启动流程

2021-01-23 15:44:39   20  举报





hadoop HDFS 中Namenode、DataNode启动流程

源码分析

流程分析

作者其他创作

大纲/内容

NNStorage

初始化DataXceiverServer 是以数据流的方式，为client提供block的数据流上传、数据流下载

blockManager.removeBlocksAssociatedTo(nodeInfo);

createNameNode创建NameNode

主要说几个重要的，后面要用的，直接通过本地方法调用的方式，RPC类似dubbo那样，不同于http的那种，通过http地址调用

只要rpc server 运行，那么就会无限制的阻塞等待，保证namenode的进程不会消失

createRpcServer创建NameNodeRpcServer

调用RPC接口

NamenodeProtocolServerSideTranslatorPB不同的Namenode通信之间的接口

edits

httpServer.start()启动

FSImage

安全模式

heartbeatManager.removeDatanode(nodeInfo);

FSNamesystem 针对磁盘操作

bpNSInfo== null

networktopology.add(node);

从conf中读取FSImage文件的路径从conf中读取edits文件的路径默认：/tmp/hadoop-用户/dfs/name他俩在一个路径，edits文件多个

fsimage

* 如果是联邦机制，也就是一组就是一个 BPOfferService ，多组就是多个BPOfferServiceHA架构里面，两个NameNode，一个active 一个stanby ，两个合在一起一组，一组就是一个 BPOfferService但是每个NameNode，就都有一个 BPServiceActor 线程BPServiceActor ：主要用来向NameNode发送请求: 注册、心跳、汇报block、以及其他的东西都在都在BPServiceActor进行处理

initBlockPool初始化BlockPool

主要判断，当前的副本数量是否满足要求每个block最小副本数 1必须满足所有block的副本率为 99.9 %也就是1000个block，必须保证999都有副本

List<DatanodeDescriptor> datanodes心跳管理的dataNode集合

FSImage ： checkpoint合并后的元数据

new BlockPoolManager初始化BlockPoolManager

添加到心跳管理

SafeModeMonitor不断的检查能否离开安全模式

checkNSEquality检查信息

DatanodeHttpServer

启动 blockManager 的一些后台线程

DatanodeRegistration

发起注册

但是这里我没看到从datanodeMap中移除

state.enterState(haContext)这里会启动响应的standby的一个线程

这个其实在NameNode这边就代表了一个DataNodeDataNode发送RPC请求时，是封装了DatanodeRegistration对象

初始化metric

核心

namesystem

refreshNamenodes

FSEditLog

handleHeartbeat

延迟调度向NameNode汇报自身的DataNode上有哪些的 block。这里主要是进行时间的计算

RPC.Server ipcServer处理RPC请求

initialize(conf)

for (BPServiceActor actor : bpServices) { actor.join(); }线程join ，注意这里的重要组件 BPServiceActor

循环遍历datanodes

loadNamesystem(conf)从`${dfs.namenode.name.dir}`目录加载fsimage与editlog，

ClientNamenodeProtocolServerSideTranslatorPBclient 和 Namenode 之间的通信调用的接口，在命令行输入的hadoop fs ****

负责管理DataNode上的block的数据

// HttpServer2 是一种通用的http服务的技术，// 如果有人给 50070端口发送请求，这里就能拦截到。然后交给Servlet进行统一请求拦截处理

过期

new FSNamesystem创建FSNamesystem ，也就是元数据信息 namespace

sendHeartbeat

heartbeatManager.addDatanode(nodeDescr)

NameNode 处理注册请求

main

createDataNode创建DataNode

NamespaceInfo

NameNodeResourceChecker

DataXceiverServer

通过 NameNode 返回的状态：active or standby 更新当前的BPServiceActor的状态，一组NameNode是有两个NameNode，一个active 一个standby ，每个NameNode是对应一个BPServiceActor保存自身NameNode的状态，如果这时候NameNode自身状态发生了转变，此时对应的 BPServiceActor 状态也会改变

管理数据节点停用。后台监视器线程定期检查正在进行退役的数据节点的状态主要是DataNode的状态

clientRpcServer.start()

重要

networktopology.add(nodeDescr);

后面的暂时不理解

versionRequest

networktopology.remove(nodeInfo);

BPOfferServiceblock pool OfferService

NameNodeHttpServer监听绑定端口、默认：50070

HeartbeatResponse resp = sendHeartBeat() 3s一次

DatanodeDescriptor

initIpcServer(conf)初始化RPC

HttpServer2

1、初始化 HttpServer22、绑定 Servlet

heartbeatManager.activate --> heartbeatThread.start()这里的 heartbeatThread 其实就是HeartbeatManager中的Monitor

namesystem.startCommonServices调用 FSNamesystem元数据启动工作

获取 DatanodeProtocolClientSideTranslatorPB （查看NameNode）

Servlet

verifyAndSetNamespaceInfo验证并设置命名空间信息

汇报 block report （默认的间隔是6个小时）

1、两个重要的时间 setLastUpdate(Time.now()); setLastUpdateMonotonic(Time.monotonicNow());2、更新DataNode的磁盘信息

初始化其他组件 FsDatasetImpl 磁盘文件读写相关

检查节点是否过期

addDatanode(nodeDescr)就是放入了几个Map中

isDatanodeDead

InterDatanodeProtocolServerSideTranslatorPBdatanode 和 datanode之间的通信接口

heartbeatThread.start()这里的 heartbeatThread 其实就是HeartbeatManager中的Monitor

ReplicationMonitor

BlockPoolManager

如果某个DataNode 十分钟都没有发送请求，就回判断这个节点可能已经是死亡状态

heartbeatManager.activate

DataNode

namenode.join()

NameNode

new DataStorage创建DataStorage

//所有心跳消息都包含以下信息：//-数据节点名称//-数据传输端口//-总容量//-剩余字节数

host2DatanodeMap.add(node);

Server

bpNSInfo == NamespaceInfo

nodeDescr

heartbeatCheck()

FsDatasetImpl 磁盘文件读写相关

其实就是一个ServerSocket处理tcp的请求

封装 DatanodeRegistration 为 DatanodeDescriptor ，

负责管理block的组件

updateHeartbeat

//HA相关暂时没看

BPServiceActor （active）

第一次启动：此时系统中肯定是没有这两个文件的，然后打开一个新的EditLog第二次启动：此时系统中有俩文件1、读取fsimage文件 2、读取edits文件 3、合并数据4、重新再写一个fsimage 文件，（旧的依旧存在）5、打开一个新的EditLog

return new DatanodeCommand[0]

RPC.Server serviceRpcServer接收datanode之间的交互信息

PendingReplicationMonitor

clientRpcServer.join()serviceRpcServer.join()

ClientDatanodeProtocolServerSideTranslatorPB处理Client和DataNode之间的管理，关闭DataNode 汇报block的接口

核心注册新的datanode

namesystem.startCommonServices启动RPC服务

50070

standbyCheckpointerStandby namenode check point

心跳检查

initDataXceiver(conf)初始化组件

resp.getCommands()看NameNode是否有需要DataNode执行什么任务

initDirectoryScanner 后台线程扫描

new NameNodeRpcServerRpcServer 主要接收DataNode的注册、心跳等等方法的直接调用请求

namesystem.startCommonServices

第二个NameNode返回了 NamespaceInfo信息，对两个NamespaceInfo校验

初始化BPOfferService

BlockManager ：集群管理

DataStorage

BPServiceActor （standby）

检查节点点是否过期：过期时间大概是 10分钟+30S

blockPoolManager.addBlockPool(bpos)在 blockPoolManager 中注册一个新的 Block pool

decomManager.activate

connectToNNAndHandshake

启动 blockManager

先分析第一次注册的情况

负责管理内存中的目录树

serviceRpcServer.start()

BlockReportProcessingThread

检查磁盘空间

否

new FSImage创建Fsimage

startDataNode

org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer#sendHeartbeat

这里两个NameNode的info信息收到后进行校验

DatanodeProtocolServerSideTranslatorPB注册Datanode、发送心跳、blockReport汇报自己的block情况等

HeartbeatManager.Monitor#run

new DatanodeCommand[cmds.size()]多个需要DataNode处理的Commadn

new NameNode(conf)

TcpPeerServer

返回需要NameNode处理的cmds

DirectoryScanner的主要任务是定期扫描磁盘上的数据块，检查磁盘上的数据块信息是否与FsDatasetImpl中保存的数据块信息一致，如果不一致则对FsDatasetImpl中的信息进行更新。注意， DirectoryScanner只会检查内存和磁盘上FINALIZED状态的数据块是否一致

createRegistration

updateActorStatesFromHeartbeat更新当前的BPServiceActor的状态

元数据主要分三块：1、内存中的目录树 2、磁盘中的edits log 3、磁盘上的FSImage

主要包含集群ID、池块id等，一些id信息名称空间节点返回NamespaceInfo，以对数据节点握手进行响应。

NameNode 处理自身的心跳检查

namesystem.loadFSImage合并edits 和 fsimage 文件

setBlockTotal()

当磁盘空间低于 100M时，就会提示磁盘不足，可以启动hasResourcesAvailable 是否有可用磁盘

FSDirectory ：负责管理内存中的目录树

其实就是初始化一个 HttpServer2

bpNSInfo = nsInfo

心跳