Hadoop3.2.1 【 HDFS 】源码分析 : Datanode 启动

灰太狼 2022-12-19 03:49 219阅读 0赞

### List of articles ###

*  一.前言
 *  二.DataNode的启动

# 一.前言 #

DataNode类封装了整个数据节点逻辑的实现。 它通过DataStorage以及FsDatasetImpl管理着数据节点存储上的所有数据块， DataNode类还会通过流式接口对客户端和其他数据节点提供读数据块、 写数据块、 复制数据块等功能。 同时DataNode类实现了InterDatanodeProtocol以及ClientDatanodeProtocol,使得数据节点可以接收来自其他数据节点以及客户端的远程RPC请求.DataNode类还会通过BlockPoolManager 对象周期性地向Namenode发送心跳、 块汇报、 增量块汇报以及缓存汇报， 同时执行Namenode发回的名字节点指令。 DataNode持有DataBlockScanner对象周期性地检查存储上的所有数据块， 以及DirectoryScanner对象验证存储上数据块和内存中数据块的一致性。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]

# 二.DataNode的启动 #

DataNode启动流程的入口方法是main()方法， 这个方法调用了secureMain()方法，secureMain()方法通过调用createDataNode()创建并启动一个DataNode实例， 然后在这个DataNode实例上调用join()方法等待DataNode停止运行.

public static void main(String args[]) {
        if (DFSUtil.parseHelpArgument(args, DataNode.USAGE, System.out, true)) {
          System.exit(0);
        }
    
        secureMain(args, null);
      }
    
    
    public static void secureMain(String args[], SecureResources resources) {
        int errorCode = 0;
        try {
          StringUtils.startupShutdownMessage(DataNode.class, args, LOG);
    
          // 构建DataNode
          DataNode datanode = createDataNode(args, null, resources);
    
          if (datanode != null) {
            //阻塞datanode
            datanode.join();
          } else {
            errorCode = 1;
          }
        } catch (Throwable e) {
          LOG.error("Exception in secureMain", e);
          terminate(1, e);
        } finally {
          // We need to terminate the process here because either shutdown was called
          // or some disk related conditions like volumes tolerated or volumes required
          // condition was not met. Also, In secure mode, control will go to Jsvc
          // and Datanode process hangs if it does not exit.
          LOG.warn("Exiting Datanode");
          terminate(errorCode);
        }
      }

createDataNode()方法首先调用静态方法instantiateDataNode()创建DataNode实例， 然后调用runDatanodeDaemon()方法启动DataNode上的各个服务。

// 实例化并启动单个datanode守护进程并等待它完成。
      // 如果这个线程被特别中断，它将停止等待。
      /** Instantiate & Start a single datanode daemon and wait for it to finish.
       *  If this thread is specifically interrupted, it will stop waiting.
       */
      @VisibleForTesting
      @InterfaceAudience.Private
      public static DataNode createDataNode(String args[], Configuration conf,
          SecureResources resources) throws IOException {
    
        //初始化构建DataNode
        DataNode dn = instantiateDataNode(args, conf, resources);
    
    
        if (dn != null) {
    
          dn.runDatanodeDaemon();
    
        }
        return dn;
      }

静态方法instantiateDataNode()首先解析DataNode的启动参数， 获取DataNode配置文件中定义的所有存储目录， 然后调用静态方法makeInstance()。 makeInstance()方法确保定义的存储目录至少有一个可用， 然后调用DataNode的构造方法创建DataNode实例。

/**
       * 实例化单个datanode对象及其安全资源
       * Instantiate a single datanode object, along with its secure resources.
       * This must be run by invoking{@link DataNode#runDatanodeDaemon()} 
       * subsequently. 
       */
      public static DataNode instantiateDataNode(String args [], Configuration conf,
          SecureResources resources) throws IOException {
    
        if (conf == null) {
          conf = new HdfsConfiguration();
        }
        
        if (args != null) {
          // parse generic hadoop options
          GenericOptionsParser hParser = new GenericOptionsParser(conf, args);
          args = hParser.getRemainingArgs();
        }
        
        if (!parseArguments(args, conf)) {
          printUsage(System.err);
          return null;
        }
    
    
        //构建存储位置对象
        Collection<StorageLocation> dataLocations = getStorageLocations(conf);
    
    
        // 构建用户组信息
        UserGroupInformation.setConfiguration(conf);
    
        // 构建授权体系 比如 kerbos
        SecurityUtil.login(conf, DFS_DATANODE_KEYTAB_FILE_KEY,
            DFS_DATANODE_KERBEROS_PRINCIPAL_KEY, getHostName(conf));
    
        // 构建实例
        return makeInstance(dataLocations, conf, resources);
      }

DataNode的构造函数在初始化了若干配置文件中定义的参数后， 调用startDataNode()方法完成DataNode的初始化操作， startDataNode()方法初始化了DataStorage对象、DataXceiverServer对象、 ShortCircuitRegistry对象， 启动了HttpInfoServer， 初始化了DataNode的IPC Server， 然后创建BlockPoolManager并加载每个块池定义的Namenode列表。

/**
       * Create the DataNode given a configuration, an array of dataDirs,
       * and a namenode proxy.
       */
      DataNode(final Configuration conf,
               final List<StorageLocation> dataDirs,
               final StorageLocationChecker storageLocationChecker,
               final SecureResources resources) throws IOException {
        super(conf);
    
        this.tracer = createTracer(conf);
        this.tracerConfigurationManager =  new TracerConfigurationManager(DATANODE_HTRACE_PREFIX, conf);
    
    
        this.fileIoProvider = new FileIoProvider(conf, this);
    
        this.blockScanner = new BlockScanner(this);
    
        this.lastDiskErrorCheck = 0;
    
        // dfs.namenode.max-num-blocks-to-log : 1000
        this.maxNumberOfBlocksToLog = conf.getLong(DFS_MAX_NUM_BLOCKS_TO_LOG_KEY,  DFS_MAX_NUM_BLOCKS_TO_LOG_DEFAULT);
    
        // dfs.block.local-path-access.user :
        this.usersWithLocalPathAccess = Arrays.asList(  conf.getTrimmedStrings(DFSConfigKeys.DFS_BLOCK_LOCAL_PATH_ACCESS_USER_KEY));
    
        // dfs.datanode.use.datanode.hostname : false
        this.connectToDnViaHostname = conf.getBoolean(
            DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME,
            DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME_DEFAULT);
    
        // dfs.permissions.superusergroup : supergroup
        this.supergroup = conf.get(DFSConfigKeys.DFS_PERMISSIONS_SUPERUSERGROUP_KEY,
            DFSConfigKeys.DFS_PERMISSIONS_SUPERUSERGROUP_DEFAULT);
    
        // dfs.permissions.enabled : true
        this.isPermissionEnabled = conf.getBoolean(
            DFSConfigKeys.DFS_PERMISSIONS_ENABLED_KEY,
            DFSConfigKeys.DFS_PERMISSIONS_ENABLED_DEFAULT);
    
        // dfs.pipeline.ecn : false
        this.pipelineSupportECN = conf.getBoolean(
            DFSConfigKeys.DFS_PIPELINE_ECN_ENABLED,
            DFSConfigKeys.DFS_PIPELINE_ECN_ENABLED_DEFAULT);
    
        // 配置文件版本 :
        confVersion = "core-" +
            conf.get("hadoop.common.configuration.version", "UNSPECIFIED") +
            ",hdfs-" +
            conf.get("hadoop.hdfs.configuration.version", "UNSPECIFIED");
    
        // 构建 DatasetVolumeChecker
        this.volumeChecker = new DatasetVolumeChecker(conf, new Timer());
    
        // 确定是否应尝试将文件描述符传递给客户端 : false
        //
        // Determine whether we should try to pass file descriptors to clients.
        if (conf.getBoolean(HdfsClientConfigKeys.Read.ShortCircuit.KEY,
                  HdfsClientConfigKeys.Read.ShortCircuit.DEFAULT)) {
    
          String reason = DomainSocket.getLoadingFailureReason();
          if (reason != null) {
            LOG.warn("File descriptor passing is disabled because {}", reason);
            this.fileDescriptorPassingDisabledReason = reason;
          } else {
            LOG.info("File descriptor passing is enabled.");
            this.fileDescriptorPassingDisabledReason = null;
          }
        } else {
          this.fileDescriptorPassingDisabledReason =
              "File descriptor passing was not configured.";
          LOG.debug(this.fileDescriptorPassingDisabledReason);
        }
    
        // 构建socketFactory
        //
        // hadoop.rpc.socket.factory.class.default: org.apache.hadoop.net.StandardSocketFactory
        this.socketFactory = NetUtils.getDefaultSocketFactory(conf);
    
        try {
    
          hostName = getHostName(conf);
          LOG.info("Configured hostname is {}", hostName);
    
          // 启动 DataNode
          startDataNode(dataDirs, resources);
    
        } catch (IOException ie) {
          shutdown();
          throw ie;
        }
    
        // dfs.datanode.network.counts.cache.max.size :  Integer.MAX_VALUE
        final int dncCacheMaxSize =
            conf.getInt(DFS_DATANODE_NETWORK_COUNTS_CACHE_MAX_SIZE_KEY,
                DFS_DATANODE_NETWORK_COUNTS_CACHE_MAX_SIZE_DEFAULT) ;
    
    
        datanodeNetworkCounts =
            CacheBuilder.newBuilder()
                .maximumSize(dncCacheMaxSize)
                .build(new CacheLoader<String, Map<String, Long>>() {
                  @Override
                  public Map<String, Long> load(String key) throws Exception {
                    final Map<String, Long> ret = new HashMap<String, Long>();
                    ret.put("networkErrors", 0L);
                    return ret;
                  }
                });
    
    
        // 从配置中获取每个OOB类型的超时值
        //
        initOOBTimeout();
    
        this.storageLocationChecker = storageLocationChecker;
      }

**startDataNode**

/**
       * This method starts the data node with the specified conf.
       * 
       * If conf's CONFIG_PROPERTY_SIMULATED property is set
       * then a simulated storage based data node is created.
       * 
       * @param dataDirectories - only for a non-simulated storage data node
       * @throws IOException
       */
      void startDataNode(List<StorageLocation> dataDirectories,
                         SecureResources resources
                         ) throws IOException {
    
        // Data Node中所有 BPs 设置全局设置
        // settings global for all BPs in the Data Node
        this.secureResources = resources;
    
        synchronized (this) {
          this.dataDirs = dataDirectories;
        }
    
        this.dnConf = new DNConf(this);
        //检查安全配置
        checkSecureConfig(dnConf, getConf(), resources);
    
        if (dnConf.maxLockedMemory > 0) {
          if (!NativeIO.POSIX.getCacheManipulator().verifyCanMlock()) {
            throw new RuntimeException(String.format(
                "Cannot start datanode because the configured max locked memory" +
                " size (%s) is greater than zero and native code is not available.",
                DFS_DATANODE_MAX_LOCKED_MEMORY_KEY));
          }
          if (Path.WINDOWS) {
            NativeIO.Windows.extendWorkingSetSize(dnConf.maxLockedMemory);
          } else {
            long ulimit = NativeIO.POSIX.getCacheManipulator().getMemlockLimit();
            if (dnConf.maxLockedMemory > ulimit) {
              throw new RuntimeException(String.format(
                "Cannot start datanode because the configured max locked memory" +
                " size (%s) of %d bytes is more than the datanode's available" +
                " RLIMIT_MEMLOCK ulimit of %d bytes.",
                DFS_DATANODE_MAX_LOCKED_MEMORY_KEY,
                dnConf.maxLockedMemory,
                ulimit));
            }
          }
        }
        LOG.info("Starting DataNode with maxLockedMemory = {}",
            dnConf.maxLockedMemory);
    
        int volFailuresTolerated = dnConf.getVolFailuresTolerated();
        int volsConfigured = dnConf.getVolsConfigured();
        if (volFailuresTolerated < MAX_VOLUME_FAILURE_TOLERATED_LIMIT
            || volFailuresTolerated >= volsConfigured) {
          throw new DiskErrorException("Invalid value configured for "
              + "dfs.datanode.failed.volumes.tolerated - " + volFailuresTolerated
              + ". Value configured is either less than -1 or >= "
              + "to the number of configured volumes (" + volsConfigured + ").");
        }
    
        //构建存储 DataStorage
        storage = new DataStorage();
        
        // global DN settings
        registerMXBean();
    
        // 开启 DataXceiverServer
        initDataXceiver();
    
        // 开启 DatanodeHttpServer
        startInfoServer();
    
        // 开启监控
        pauseMonitor = new JvmPauseMonitor();
        pauseMonitor.init(getConf());
        pauseMonitor.start();
      
        // BlockPoolTokenSecretManager is required to create ipc server.
        this.blockPoolTokenSecretManager = new BlockPoolTokenSecretManager();
    
        // Login is done by now. Set the DN user name.
        dnUserName = UserGroupInformation.getCurrentUser().getUserName();
        LOG.info("dnUserName = {}", dnUserName);
        LOG.info("supergroup = {}", supergroup);
        initIpcServer();
    
        metrics = DataNodeMetrics.create(getConf(), getDisplayName());
        peerMetrics = dnConf.peerStatsEnabled ?
            DataNodePeerMetrics.create(getDisplayName(), getConf()) : null;
        metrics.getJvmMetrics().setPauseMonitor(pauseMonitor);
    
        // ErasureCodingWorker
        ecWorker = new ErasureCodingWorker(getConf(), this);
        // block恢复Worker
        blockRecoveryWorker = new BlockRecoveryWorker(this);
    
        // 构建BlockPoolManager
        blockPoolManager = new BlockPoolManager(this);
    
        // 刷新namenode
        blockPoolManager.refreshNamenodes(getConf());
    
        // Create the ReadaheadPool from the DataNode context so we can
        // exit without having to explicitly shutdown its thread pool.
        readaheadPool = ReadaheadPool.getInstance();
    
        saslClient = new SaslDataTransferClient(dnConf.getConf(),
            dnConf.saslPropsResolver, dnConf.trustedChannelResolver);
    
        saslServer = new SaslDataTransferServer(dnConf, blockPoolTokenSecretManager);
    
        startMetricsLogger();
    
        if (dnConf.diskStatsEnabled) {
          diskMetrics = new DataNodeDiskMetrics(this,
              dnConf.outliersReportIntervalMs);
        }
      }

成功初始化DataNode对象之后， 就需要调用runDatanodeDaemon()方法启动DataNode的服务了。 runDatanodeDaemon()方法启动了blockPoolManager管理的所有线程， 启动了DataXceiverServer线程， 最后启动了DataNode的IPC Server。

/**
       * 开启各种服务
       *  Start a single datanode daemon and wait for it to finish.
       *  If this thread is specifically interrupted, it will stop waiting.
       */
      public void runDatanodeDaemon() throws IOException {
    
    
        blockPoolManager.startAll();
    
        // start dataXceiveServer
        dataXceiverServer.start();
        if (localDataXceiverServer != null) {
          localDataXceiverServer.start();
        }
        ipcServer.setTracer(tracer);
        ipcServer.start();
        startPlugins(getConf());
    
      }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/fbaab406842147cbb554b4944e4a1159.png

发表评论取消回复

表情：

评论列表（有 0 条评论，219人围观）

还没有评论，来说两句吧...

相关阅读

相关 HDFS源码分析——DataNode启动流程

DataNode启动流程 1. 概述 2. 解析 secureMain→createDataNode→instantiateDataNode→ma

╰半夏微凉°/ 2024年04月03日 10:16/ 0 赞/ 83 阅读

相关 HDFS源码分析——NameNode启动流程

NameNode启动流程准备工作 1. 概述 2. 解析 main→createNameNode→new NameNode→initial

川长思鸟来/ 2024年04月03日 08:56/ 0 赞/ 84 阅读

相关【Hadoop】Datanode启动失败

> 启动hadoop后，用jps命令查看，始终没有DataNode。 > > ![20200311235140757.png][] 解决办法：首先关闭hado

亦凉/ 2023年07月13日 12:17/ 0 赞/ 159 阅读

相关 hadoop 源码分析（一）HDFS nameNode 之 httpServer启动源码分析

一、HDFS概述基于hadoop2.6 HDFS被设计能够运行在通用硬件上、提供流式数据操作、能够处理超大文件的分布式文件系统。特点：高容错和高吞吐量、易扩展

谁践踏了优雅/ 2023年01月03日 04:58/ 0 赞/ 294 阅读

相关 Hadoop3.2.1 【 HDFS 】源码分析 : Datanode升级、回滚、提交

List of articles 一.前言一.doUpgrade() 三.doRollback() 四.doFinalize() 一.前言

本是古典何须时尚/ 2022年12月20日 02:25/ 0 赞/ 241 阅读

相关 Hadoop3.2.1 【 HDFS 】源码分析 : Datanode 启动

List of articles 一.前言二.DataNode的启动一.前言 DataNode类封装了整个数据节点逻辑的实现。它通过DataSt

灰太狼/ 2022年12月19日 03:49/ 0 赞/ 220 阅读

相关 Hadoop3.2.1 【 HDFS 】源码分析 : Datanode 存储[二]

List of articles 一. Storage 二.DataStorage 三. BlockPoolSliceStorage 一. Stor

墨蓝/ 2022年11月22日 10:20/ 0 赞/ 158 阅读

相关 Hadoop3.2.1 【 HDFS 】源码分析 : Datanode 存储[一]

List of articles 一. 前言二.Datanode磁盘存储结构三.DataStorage实现 3.1. Storage类继

ゝ一纸荒年。/ 2022年11月22日 00:07/ 0 赞/ 284 阅读

相关 Hadoop3.2.1 【 HDFS 】源码分析 : Datanode 概述

一.概述 Datanode以存储数据块（Block）的形式保存HDFS文件，同时Datanode还会响应HDFS客户端读、写数据块的请求。 Datanode会周期性

Love The Way You Lie/ 2022年11月21日 15:22/ 0 赞/ 262 阅读

相关 Hadoop中启动hdfs时DataNode没有启动

是刚开始学习Hadoop，这个问题困扰了我一个多小时，为了以后少犯这种错误，也为了别人能少走歪路，我记录下来我的解决方法，可能不是最快捷的，但是这是我摸索出来的。 hd...

朱雀/ 2020年10月25日 08:10/ 0 赞/ 1034 阅读