DataNode启动流程源码分析

程序员文章站 2022-07-11 23:49:27

...

我们都知道在Hadoop hdfs文件系统中，Datanode是负责hdfs文件对应的数据块存储管理的组件，其会在启动时向NameNode汇报其上拥有的数据块，以及周期性心跳并接收来自NameNode下发的对应的数据块指令等等；其DataNode和NameNode的大致交互流程包括：

Datanode启动时的握手、注册流程；
数据块汇报以及增量汇报等流程；
周期性心跳流程(当前数据节点负载、接受来自NameNode的指令)；

DataNode启动流程源码分析

先来看下DataNode的基本结构逻辑图，可以按照数据层、逻辑层、服务层划分模块如下：

DataNode启动流程源码分析

数据层：将DataNode中负责数据块存储和管理数据块操作的部分抽象成数据层，它主要包括2部分：

DataStorage(数据块存储)：数据块存储主要是管理DataNode磁盘存储空间以及磁盘存储空间的生命周期。说的直白一点就是DataStorage这个类主要负责管理数据存储文件信息，比如current，detach,finalized等等文件。BlockPoolSliceStorage可以用来管理DataNode每一个单独的块池，DataStorage会持有这个BlockPoolSliceStorage对象引用，并通过这个引用来管理DataNode的块池。
FSDataset(文件系统数据集)：FSDataset抽象了DataNode对数据块管理的操作，比如创建数据块，维护数据块文件等。我们知道每一个DataNode都可以配置多个不同类型的存储目录来保存数据，所以HDFS定义了FSVolumeImpl来管理DataNode上单个存储目录上保存的数据块，同时定义了FSVolumeList来维护DataNode上所有FSVolumeImpl对象的引用。FSDataSet会通过FSVolumeList提供的管理功能来管理DataNode上存储的目录保存的数据块

逻辑层：DataNode基于数据层会执行很多HDFS逻辑处理，比如向NameNode汇报数据块状态，发送心跳，扫描损坏的数据块等，我们将HDFS执行这些逻辑的部分抽象成逻辑层。逻辑层主要包括三个模块：

BlockPoolManager：BlockPoolManager是管理所有BlockPool的接口类，在HDFS Federation机制下，我们在集群可以创建多个NameSpace，每一个NameSpace都对应着一个BlockPool,一个BlockPoolManager会持有一个BPOfferService对象，用于管理DataNode单个BlockPool，我们知道，如果引入HA机制， NameNode就会有ActiveNM 和 StandbyNM。所以每一个BPOfferService又会持有2个BPServiceActor对象，每一个BPServiceActor对应于命名空间里的一个Name Node，该对象负责向NameNode发送心跳报告，数据块汇报，缓存汇报等
DataBlockScanner：一个周期性扫描每一个数据块并检查数据块校验是否正常的一个线程
DirectoryScanner：周期性扫描磁盘数据块，对比内存中元数据与实际磁盘存储数据块的差异，并根据差异更新内存元数据，使得与磁盘保存一致

服务层：主要用于客户端或者其他节点和DataNode通信，以及访问DataNode状态等功能，主要包括三个模块：

HttpServer: 对外提供http服务
IpcServer：RPC服务端，响应来自客户端，NameNode和其他DataNode的rpc请求
DataXceiverServer：输出传输服务端，响应来自客户端以及其他Data Node的流式接口请求

接下来逐步分析Datanode的初始化启动流程以及其对外提供的服务：

初始化构造时参数设置，并构造内部服务组件：
1. 获取节点的名字和名字节点的地址，读入运行时配置项
2. 构造DataNodeRegistration对象
3. RPC调用名字节点上的handshake()方法，建立到名字节点的IPC连接
4. 执行存储空间状态恢复，构造数据节点的FSDataSet对象
5. 创建流式接口服务器：DataXceiverServer，数据块扫描器DataBlockScanner等
6. 创建数据节点上的Http服务器(主要用于界面展示数据节点运行状态)
7. 创建数据节点IPC服务器，对外提供ClientDataNodeProtocol和InterDataNodeProtocol接口协议服务
DataNode.register()向NameNode注册当前数据节点，并执行数据节点的服务主线程
DataNode.run()方法-->循环执行offerService()：
1. 发送到NameNode的心跳，并执行可能的名字节点指令
2. 通过BlockReceived()方法向NameNode上报数据节点上接收到的数据块
3. 根据远程接口DataNodeProtocol.blockReport()向NameNode报告数据节点目前保存的数据块信息
4. 启动数据块扫描器DataBlockScanner和DirectoryScanner周期性检查block和dir是否可用损坏等

其启动的详细源码在DataNode.main()中，其会调用createDataNode()方法来创建并启动一个DataNode实例；之后便在该DataNode实例上调用.join()方法阻塞等待DataNode停止运行。其基本的createDataNode()创建启动DataNode实例方法如下：

public static DataNode createDataNode(String args[], Configuration conf,
    SecureResources resources) throws IOException {
  // 完成大部分初始化的工作，并启动部分工作线程
  DataNode dn = instantiateDataNode(args, conf, resources);
  if (dn != null) {
    // 启动剩余工作线程
    dn.runDatanodeDaemon();
  }
  return dn;
}

1、instantiateDataNode()：

从conf文件中获取到数据存储路径（dfs.datanode.data.dir）
使用三个参数（数据存储路径、配置文件、SecureResources）去实例化Datanode

public static DataNode instantiateDataNode(String args [], Configuration conf,
    SecureResources resources) throws IOException {
  if (conf == null)
    conf = new HdfsConfiguration();
  
  // ...... 参数检查等
  
  Collection<StorageLocation> dataLocations = getStorageLocations(conf);
  UserGroupInformation.setConfiguration(conf);
  SecurityUtil.login(conf, DFS_DATANODE_KEYTAB_FILE_KEY,
      DFS_DATANODE_KERBEROS_PRINCIPAL_KEY);
  return makeInstance(dataLocations, conf, resources);
}

之后便从DataNode.makeInstance()开始创建DataNode实例；makeInstance方法主要的作用：

获取客户端校验类，拿到存储数据目录的权限并传入磁盘检测对象进行磁盘检测，调用checkStorageLocations方法利用磁盘检测对象进行磁盘目录的检测，返回可用磁盘目录列表
1. 声明一个集合，用来存储可用目录列表
2. 遍历数据目录,注意这里就是串行的方式
3. 利用磁盘检测对象进行磁盘目录的检测：校验目录的读写执行权限，如果目录不存在，则创建目录并给予700权限
4. 检测完毕没有抛出异常，则说明目录可用，加入到可用列表
5. 如果出现IO异常，说明此磁盘目录不可用，加入到目录中
6. 如果可用目录数量为0，表明所有的目录都不可用
7. 最终，返回可用的磁盘目录列表
使用构造函数创建datanode对象：
1. 根据configuration，初始化一些成员变量
2. 给出一个配置、一个dataDirs数组和一个Namenode代理，创建DataNode。
3. 调用startDataNode(conf, dataDirs, resources)方法启动datanode如下：

void startDataNode(Configuration conf, 
                   List<StorageLocation> dataDirs,
                   SecureResources resources
                   ) throws IOException {
  // ......参数设置

  // 初始化DataStorage
  storage = new DataStorage();
  
  // global DN settings
  registerMXBean(); // 注册JMX
  initDataXceiver(conf); // 初始化DataXceiverServer（流式通信），DataNode#runDatanodeDaemon()中启动
  startInfoServer(conf); // 启动InfoServer（Web UI）
 
  pauseMonitor = new JvmPauseMonitor(conf); // 启动JVMPauseMonitor（反向监控JVM情况，可通过JMX查询）
  pauseMonitor.start();

  // ......
  
  // 初始化IpcServer（RPC通信），DataNode#runDatanodeDaemon()中启动
  initIpcServer(conf);

  metrics = DataNodeMetrics.create(conf, getDisplayName());
  metrics.getJvmMetrics().setPauseMonitor(pauseMonitor);
  
  // 按照namespace（nameservice）、namenode的二级结构进行初始化BlockPoolManager
  blockPoolManager = new BlockPoolManager(this);
  blockPoolManager.refreshNamenodes(conf);
  
  // ......
}

其具体的startDataNode(conf, dataDirs, resources)启动datanode的源码分析如下：

实例化管理磁盘目录的DataStorage：
1. DataStorage：管理与组织磁盘存储目录，如current,previous,detach,tmp等；在DataNode数据目录，你可以看到一些current tmp,rbw或者finalized文件夹
2. FsDatasetImpl：管理组织数据块和元数据文件
启动DataXceiverServer流式接口服务器组件：DataXceiverServer是数据节点DataNode上一个用于接收数据块读写请求的后台工作线程，其会为每个数据块的读写请求创建一个单独的线程去处理；
1. 创建构造DataXceiverServer需要的TcpPeerServer实例tcpPeerServer，它内部封装了ServerSocket，是DataXceiverServer功能实现的最主要依托；
2. 从tcpPeerServer中获取Socket地址InetSocketAddress，赋值给DataNode成员变量streamingAddr
3. 然后构造DataXceiverServer实例xserver，传入tcpPeerServer；
4. 构造dataXceiverServer守护线程，并将xserver加入之前创建的线程组threadGroup；
5. 将线程组里的所有线程设置为设置为守护线程，方便虚拟机退出时自动销毁；
启动Http服务和RPC服务：
1. startInfoServer()主要用于启动DataNode对外提供的web UI信息
2. initIpcServer()初始化IpcServer（RPC通信），用来构建datanode上的rpc服务，主要包括两个服务：ClientDataNodeProtocol和InterDataNodeProtocol协议服务
构造BlockPoolManager组件：其抽象了datanode提供的数据块存储服务，每个DataNode都有一个BlockPoolManager实例，并通过blockPoolManager.refreshNamenodes(conf)从配置文件中获取该datanode相关的namenode信息，并为每个namespace创建对应的BPOfferService服务，然后向其namenode发送注册和心跳信息。

2、dn.runDatanodeDaemon()：

在构造完毕DataNode上所拥有的组件及服务后，便可以调用dn.runDatanodeDaemon()来启动DataNode上组件对应的服务：

public void runDatanodeDaemon() throws IOException {
  blockPoolManager.startAll();

  // start dataXceiveServer
  dataXceiverServer.start();
  if (localDataXceiverServer != null) {
    localDataXceiverServer.start();
  }
  ipcServer.start();
  startPlugins(conf);
}

接下来详细分析下DataNode向NameNode注册以及其心跳机制如下：

从上文DataNode构造初始化过程中可以知道，其在datanode的构造方法里，初始化了BlockPoolManager实例对象，并通过blockPoolManager.refreshNamenodes(conf)；从配置文件中获取该datanode相关的namenode信息，然后向其发送注册和心跳信息。其具体会调用BlockPoolManager里面的startAll()方法，通过startAll方法，会将datanode上面的所有BPOfferService启动：

1、BlockPoolManager#doRefreshNamenodes()：

通过BPOfferService bpos = createBPOS(addrs)为每个namespace创建对应的BPOfferService（包括每个namenode对应的BPServiceActor）
然后通过BlockPoolManager#startAll()启动所有BPOfferService（实际是启动所有 BPServiceActor）

2、BPServiceActor线程的启动：

存储结构初始化、启动DataBlockScanner、DirectoryScanner等工作线程
向namonode握手、注册、数据块上报、心跳

BPServiceActor线程：

DataNode中实际与NameNode进行通信的正是BPServiceActor线程。

@Override
public void run() {
  LOG.info(this + " starting to offer service");
  try {
    while (true) {
      // init stuff
      try {
        // 与namonode握手，注册
        connectToNNAndHandshake();
        break;
      } catch (IOException ioe) {
        // 大部分握手失败的情况都需要重试，除非抛出了非IOException异常或datanode关闭
      }
    }
    
    runningState = RunningState.RUNNING;

    while (shouldRun()) {
      try {
        // BPServiceActor提供的服务
        offerService();
      } catch (Exception ex) {
        // 不管抛出任何异常，都持续提供服务（包括心跳、数据块汇报等），直到datanode关闭
      }
    }
    runningState = RunningState.EXITED;
  } catch (Throwable ex) {
    // 资源清理
  }
}

1、BPServiceActor#connectToNNAndHandshake()：其会与NameNode进行握手并初始化DataNode上该命名空间对应块池(BlockPool)的存储，然后在该NameNode上注册当前DataNode。

额外需要注意的是在BPOfferService#verifyAndSetNamespaceInfo()中：

如果是第一次连接namenode(也是第一次连接namespace)，则需要以BPOfferService为单位初始化blockpool（块池）
在initBlockPool()方法中，其会初始化块池对应的DataStorage、初始化FsDatasetImpl对象；并初始化启动DataBlockScanner数据块扫描线程、和DirectoryScanners目录检测线程

private void connectToNNAndHandshake() throws IOException {
  // get NN proxy  创建NameNode的rpc客户端代理
  bpNamenode = dn.connectToNN(nnAddr);

  // 先通过第一次握手获得namespace的信息
  NamespaceInfo nsInfo = retrieveNamespaceInfo();
  
  // 然后验证并初始化该datanode上的BlockPool
  bpos.verifyAndSetNamespaceInfo(nsInfo);
  
  // 最后，通过第二次握手向各namespace注册自己
  register();
}

2、offerService()：offerService()方法是BPServiceActor的主循环方法，它向NameNode发送心跳、数据块汇报(块汇报、缓存汇报以及增量汇报)。其基本执行流程如下：

DataNode启动流程源码分析

private void offerService() throws Exception {
  while (shouldRun()) {
    try {
      final long startTime = now();
      // Every so often, send heartbeat or block-report
      // 发送心跳信息(携带该DN上的负载信息)
      if (startTime - lastHeartbeat >= dnConf.heartBeatInterval) {
        //
        // All heartbeat messages include following info:
        // -- Datanode name
        // -- data transfer port
        // -- Total capacity
        // -- Bytes remaining
        //
        lastHeartbeat = startTime;
        if (!dn.areHeartbeatsDisabledForTests()) {
          // 心跳信息的发送
          HeartbeatResponse resp = sendHeartBeat();
          assert resp != null;
          dn.getMetrics().addHeartbeat(now() - startTime);
           
          // ......
          // 处理NameNode返回的执行指令
          long startProcessCommands = now();
          if (!processCommand(resp.getCommands()))
            continue;
          long endProcessCommands = now();
          if (endProcessCommands - startProcessCommands > 2000) {
            LOG.info("Took " + (endProcessCommands - startProcessCommands)
                + "ms to process " + resp.getCommands().length
                + " commands from NN");
          }
        }
      }
      if (sendImmediateIBR ||
          (startTime - lastDeletedReport > dnConf.deleteReportInterval)) {
        reportReceivedDeletedBlocks(); // 增量块汇报
        lastDeletedReport = startTime;
      }

      List<DatanodeCommand> cmds = blockReport(); // 全量块汇报
      processCommand(cmds == null ? null : cmds.toArray(new DatanodeCommand[cmds.size()]));

      DatanodeCommand cmd = cacheReport(); // 缓存数据块汇报
      processCommand(new DatanodeCommand[]{ cmd });

      // Now safe to start scanning the block pool.
      // If it has already been started, this is a no-op.
      if (dn.blockScanner != null) {
        dn.blockScanner.addBlockPool(bpos.getBlockPoolId()); // 启动数据块扫描
      }

      //
      // There is no work to do;  sleep until hearbeat timer elapses, 
      // or work arrives, and then iterate again.
      // 睡眠等待，直到下一个心跳周期或者被唤醒
      long waitTime = dnConf.heartBeatInterval - 
      (Time.now() - lastHeartbeat);
      synchronized(pendingIncrementalBRperStorage) {
        if (waitTime > 0 && !sendImmediateIBR) {
          try {
            pendingIncrementalBRperStorage.wait(waitTime);
          } catch (InterruptedException ie) {
            LOG.warn("BPOfferService for " + this + " interrupted");
          }
        }
      } // synchronized
    } catch(RemoteException re) {
      // 异常处理
    }
  } // while (shouldRun())
} // offerService

接下来简单分析一下sendHeartBeat()发送心跳、processCommand()名字节点指令处理、blockReport()全量块汇报等方法(其余涉及到的方法请自行分析)：

sendHeartBeat()发送心跳：

@Override
public HeartbeatResponse sendHeartbeat(DatanodeRegistration registration,
    StorageReport[] reports, long cacheCapacity, long cacheUsed,
        int xmitsInProgress, int xceiverCount, int failedVolumes)
            throws IOException {
  // 构造心跳信息
  HeartbeatRequestProto.Builder builder = HeartbeatRequestProto.newBuilder()
      .setRegistration(PBHelper.convert(registration))
      .setXmitsInProgress(xmitsInProgress).setXceiverCount(xceiverCount)
      .setFailedVolumes(failedVolumes);
  builder.addAllReports(PBHelper.convertStorageReports(reports));
  
  // 调用NameNode RPC接口 发送心跳
  HeartbeatResponseProto resp;
  try {
    resp = rpcProxy.sendHeartbeat(NULL_CONTROLLER, builder.build());
  } catch (ServiceException se) {
    throw ProtobufHelper.getRemoteException(se);
  }
  
  // 获取NameNode返回的命令
  DatanodeCommand[] cmds = new DatanodeCommand[resp.getCmdsList().size()];
  // ......
  return new HeartbeatResponse(cmds, PBHelper.convert(resp.getHaStatus()),
      rollingUpdateStatus);
}

processCommand()名字节点指令处理：其最终会调用BPOfferService#processCommandFromActive(cmd, actor)；

private boolean processCommandFromActive(DatanodeCommand cmd,
    BPServiceActor actor) throws IOException {
  // ......
  
  switch(cmd.getAction()) {
  case DatanodeProtocol.DNA_TRANSFER: // 向其他DN传输block
    // Send a copy of a block to another datanode
    dn.transferBlocks(bcmd.getBlockPoolId(), bcmd.getBlocks(),
        bcmd.getTargets(), bcmd.getTargetStorageTypes());
    dn.metrics.incrBlocksReplicated(bcmd.getBlocks().length);
    break;
  case DatanodeProtocol.DNA_INVALIDATE: // 删除数据块操作
    Block toDelete[] = bcmd.getBlocks();
    try {
      if (dn.blockScanner != null) {
        dn.blockScanner.deleteBlocks(bcmd.getBlockPoolId(), toDelete);
      }
      // using global fsdataset
      dn.getFSDataset().invalidate(bcmd.getBlockPoolId(), toDelete);
    } catch(IOException e) {
      // Exceptions caught here are not expected to be disk-related.
      throw e;
    }
    dn.metrics.incrBlocksRemoved(toDelete.length);
    break;
  case DatanodeProtocol.DNA_CACHE:
    // ...... 缓存相关操作
  case DatanodeProtocol.DNA_UNCACHE:
    // ...... 缓存相关操作
  case DatanodeProtocol.DNA_SHUTDOWN: 
    // ...... 关闭DN
  case DatanodeProtocol.DNA_FINALIZE: // 提交升级
    String bp = ((FinalizeCommand) cmd).getBlockPoolId();
    dn.finalizeUpgradeForPool(bp);
    break;
  case DatanodeProtocol.DNA_RECOVERBLOCK: // 块恢复
    String who = "NameNode at " + actor.getNNSocketAddress();
    dn.recoverBlocks(who, ((BlockRecoveryCommand)cmd).getRecoveringBlocks());
    break;
  case DatanodeProtocol.DNA_ACCESSKEYUPDATE:
    LOG.info("DatanodeCommand action: DNA_ACCESSKEYUPDATE");
    if (dn.isBlockTokenEnabled) {
      dn.blockPoolTokenSecretManager.addKeys(
          getBlockPoolId(), 
          ((KeyUpdateCommand) cmd).getExportedKeys());
    }
    break;
  case DatanodeProtocol.DNA_BALANCERBANDWIDTHUPDATE: // 节流器控制
    LOG.info("DatanodeCommand action: DNA_BALANCERBANDWIDTHUPDATE");
    long bandwidth =
               ((BalancerBandwidthCommand) cmd).getBalancerBandwidthValue();
    if (bandwidth > 0) {
      DataXceiverServer dxcs =
                   (DataXceiverServer) dn.dataXceiverServer.getRunnable();
      LOG.info("Updating balance throttler bandwidth from "
          + dxcs.balanceThrottler.getBandwidth() + " bytes/s "
          + "to: " + bandwidth + " bytes/s.");
      dxcs.balanceThrottler.setBandwidth(bandwidth);
    }
    break;
  default:
    LOG.warn("Unknown DatanodeCommand action: " + cmd.getAction());
  }
  return true;
}

blockReport()全量块汇报：

List<DatanodeCommand> blockReport() throws IOException {
  // Flush any block information that precedes the block report. Otherwise
  // we have a chance that we will miss the delHint information
  // or we will report an RBW replica after the BlockReport already reports
  // a FINALIZED one.
  reportReceivedDeletedBlocks(); // 增量块汇报
  lastDeletedReport = startTime;

  // 获取所有数据块的状态以及信息
  // 从数据节点DataNode根据线程对应块池ID获取数据块汇报集合perVolumeBlockLists，
  // key为数据节点存储DatanodeStorage，value为数据节点存储所包含的Long类数据块数组BlockListAsLongs
  long brCreateStartTime = now();
  Map<DatanodeStorage, BlockListAsLongs> perVolumeBlockLists =
      dn.getFSDataset().getBlockReports(bpos.getBlockPoolId());

  // Convert the reports to the format expected by the NN.
  // ......格式转换
  
  // Send the reports to the NN.
  int numReportsSent = 0;
  int numRPCs = 0;
  boolean success = false;
  long brSendStartTime = now();
  long reportId = generateUniqueBlockReportId();
  // 根据数据块总数目判断是否需要多次发送消息
  try {
    if (totalBlockCount < dnConf.blockReportSplitThreshold) {
      // Below split threshold, send all reports in a single message.
      // 通过NameNode代理bpNamenode的blockReport()方法向NameNode发送数据块汇报信息
      DatanodeCommand cmd = bpNamenode.blockReport(
          bpRegistration, bpos.getBlockPoolId(), reports,
            new BlockReportContext(1, 0, reportId));
      numRPCs = 1;
      numReportsSent = reports.length;
      if (cmd != null) {
        cmds.add(cmd);
      }
    } else {
      // Send one block report per message.
      for (int r = 0; r < reports.length; r++) {
        StorageBlockReport singleReport[] = { reports[r] };
        DatanodeCommand cmd = bpNamenode.blockReport(
            bpRegistration, bpos.getBlockPoolId(), singleReport,
            new BlockReportContext(reports.length, r, reportId));
        numReportsSent++;
        numRPCs++;
        if (cmd != null) {
          cmds.add(cmd);
        }
      }
    }
    success = true;
  } finally {
    // Log the block report processing stats from Datanode perspective
  }
  // 调用scheduleNextBlockReport()方法，调度下一次数据块汇报；并返回命令cmds
  scheduleNextBlockReport(startTime);
  return cmds.size() == 0 ? null : cmds;
}

相关标签： Hadoop分析与理解 hadoop datanode 注册心跳 heartbeat

上一篇： Flume日志收集

下一篇：鄂尔多斯人均有3到4套房斥资60亿新城成泡沫最佳标本

DataNode启动流程源码分析

1、instantiateDataNode()：

2、dn.runDatanodeDaemon()：

BPServiceActor线程：

Android 2.3 拨号上网流程从源码角度进行分析

Netty源码分析（三）----- 服务端启动源码分析

Tomcat源码分析三：Tomcat启动加载过程（一）的源码解析

104 - kube-scheduler源码分析 - predicate整体流程

详解Spring IOC 容器启动流程分析

MapReduce之Job提交流程源码和切片源码分析

Tomcat源码分析（六）----- Tomcat 启动过程(一)

Fpm启动机制及流程分析———详细

Android10.0-多用户切换流程源码分析

Flink的Job启动Driver端(源码分析)

DataNode启动流程源码分析

1、instantiateDataNode()：

2、dn.runDatanodeDaemon()：

BPServiceActor线程：

Android 2.3 拨号上网流程从源码角度进行分析

Netty源码分析 （三）----- 服务端启动源码分析

Tomcat源码分析三：Tomcat启动加载过程（一）的源码解析

104 - kube-scheduler源码分析 - predicate整体流程

详解Spring IOC 容器启动流程分析

MapReduce之Job提交流程源码和切片源码分析

Tomcat源码分析 （六）----- Tomcat 启动过程(一)

Fpm启动机制及流程分析———详细

Android10.0-多用户切换流程源码分析

Flink的Job启动Driver端(源码分析)

Netty源码分析（三）----- 服务端启动源码分析

Tomcat源码分析（六）----- Tomcat 启动过程(一)