Hadoop3.2.1 【 HDFS 】源码分析 : 文件系统数据集 [二]

淡淡的烟草味﹌ 2022-12-20 05:54 152阅读 0赞

### List of articles ###

*  一. 前言
 *  二.FsVolumeImpl实现
 *   *  2.1. FsVolumeSpi接口定义
     *  2.2. FsVolumeImpl的字段和方法
 *  三.FsVolumeList实现

# 一. 前言 #

Datanode可以配置多个存储目录保存数据块文件（这里要注意， Datanode的多个存储目录存储的数据块并不相同， 并且不同的存储目录可以是异构的， 这样的设计可以提高数据块IO的吞吐率） ， 所以Datanode将底层数据块的管理抽象为多个层次， 并定义不同的类  
来实现各个层次上数据块的管理。

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

■ BlockPoolSlice： 管理一个指定块池在一个指定存储目录下的所有数据块。 由于Datanode可以定义多个存储目录， 所以块池的数据块会分布在多个存储目录下。一个块池会拥有多个BlockPoolSlice对象， 这个块池对应的所有BlockPoolSlice对象共同管理块池的所有数据块。
    ■ FsVolumelmpl： 管理Datanode一个存储目录下的所有数据块。 由于一个存储目录可以存储多个块池的数据块， 所以FsVolumelmpl会持有这个存储目录中保存的所有块池的BlockPoolSlice对象。
    ■ FsVolumeList： Datanode可以定义多个存储目录， 每个存储目录下的数据块是使用一个FsVolumelmpl对象管理的， 所以Datanode定义了FsVolumeList类保存Datanode上所有的FsVolumelmpl对象， FsVolumeList对FsDatasetlmpl提供类似磁盘的服务。

# 二.FsVolumeImpl实现 #

Datanode可以配置多个存储目录存储数据块文件。  
每一个存储目录下的数据块管理功能都由一个单独的FsVolumeImpl对象负责。 而每一个存储目录又可以保存多个块池的数据块， 其中每一个块池数据块的管理都由一个BlockPoolSlice对象负责， 因此FsVolumeImpl对象会持有多个BlockPoolSlice对象的引用。

## 2.1. FsVolumeSpi接口定义 ##

FsVolumeSpi是FsVolumeImpl的根接口,定义了如下几个方法:

■ getStorageID()： 获取当前存储目录的StorageUuId。
    ■ getBlockPoolList()： 获取当前存储目录下的块池列表， 这里返回的是当前存储目录下保存的所有块池的块池id(blockPoolId） 。
    ■ getAvailabe()： 获取当前存储目录的可用空间。
    ■ getBasePath()： 获取当前存储目录的根路径。
    ■ getStorageType()： 获取当前存储目录的存储类型， 存储目录类型可以是内存、SSD或者磁盘。
    ■ reserveSpaceForReplica()： 对RBW状态的数据块预留磁盘空间， 这样写入数据块时就不会出现磁盘空间不足的情况了。
    ■ releaseReservedSpace()： 释放预留的空间。
    ■ isTransientStorage()： 判断当前卷是否有与之对应的持久化磁盘。
    其他略........

## 2.2. FsVolumeImpl的字段和方法 ##

**FsVolumeImpl字段:**

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

■ dataset： 这个字段主要用于加锁操作。  
■ storageID： 当前存储目录对应的StorageDirectory的storageID。  
■ currentDir： 当前存储目录下current文件夹的引用。  
■ usage： 当前存储目录的磁盘使用情况。  
■ reserved： 当前存储目录的预留磁盘空间大小。  
■ bpSlices ： 当前FsVolumeImpl下所有BlockPoolSlice的引用， 是blockPoolId->BlockPoolSlice的映射。  
■ cacheExecutor： 线程池， 用来处理添加到缓存中的新的数据块。  
■ reservedForRbw： 预留的磁盘空间

//  这个字段主要用于加锁操作
      private final FsDatasetImpl dataset;
    
      // 当前存储目录对应的StorageDirectory的storageID。
      private final String storageID;
      private final StorageType storageType;
    
      // 当前FsVolumeImpl下所有BlockPoolSlice的引用， 是blockPoolId->BlockPoolSlice的映射。
      private final Map<String, BlockPoolSlice> bpSlices  = new ConcurrentHashMap<String, BlockPoolSlice>();
    
      // Refers to the base StorageLocation used to construct this volume
      // (i.e., does not include STORAGE_DIR_CURRENT in
      // <location>/STORAGE_DIR_CURRENT/)
      private final StorageLocation storageLocation;
    
      // 当前存储目录下current文件夹的引用。
      private final File currentDir;    // <StorageDirectory>/current
    
      // 当前存储目录的磁盘使用情况。
      private final DF usage;
      private final ReservedSpaceCalculator reserved;
      private CloseableReferenceCount reference = new CloseableReferenceCount();
    
      // Disk space reserved for blocks (RBW or Re-replicating) open for write.
      private AtomicLong reservedForReplicas;
    
      // 当前存储目录的预留磁盘空间大小。
      private long recentReserved = 0;
      private final Configuration conf;
      // Capacity configured. This is useful when we want to
      // limit the visible capacity for tests. If negative, then we just
      // query from the filesystem.
      protected volatile long configuredCapacity;
      private final FileIoProvider fileIoProvider;
      private final DataNodeVolumeMetrics metrics;
    
      /**
       * 线程池， 用来处理添加到缓存中的新的数据块。
       *
       * Per-volume worker pool that processes new blocks to cache.
       * The maximum number of workers per volume is bounded (configurable via
       * dfs.datanode.fsdatasetcache.max.threads.per.volume) to limit resource
       * contention.
       */
    
      protected ThreadPoolExecutor cacheExecutor;

**FsVolumeImpl定义的方法**

FsVolumeImpl的方法可以分为三类  
■ 获取存储目录的存储状态： 包括getDfsUsed()、 getBlockPoolUsed()、getCapactiy()、 getAvailable()、 getReserved()等方法。 这些方法都用于获取存储目录的磁盘使用状态， 底层基本都是通过调用BlockPoolSlice对应方法获得返回值的。

getDfsUsed()方法调用了这个存储目录管理的所有BlockPoolSlice对象的getDfsUsed()方法， 并将这些值累加并返回。  
代码如下：

public long getDfsUsed() throws IOException {
        long dfsUsed = 0;
        for(BlockPoolSlice s : bpSlices.values()) {
          dfsUsed += s.getDfsUsed();
        }
        return dfsUsed;
      }

■ 获取存储目录的子目录： 包括getCurrentDir()、 getRbwDir()以及getLazyPersistDir()三个方法， 用于获取存储目录下的特定子目录。 其中getCurrentDir()用于获取存储目录下的current子目录， 方法底层是直接返回了FsVolumeImpl.currentDir字段； 而getRbwDir()getLazyPersistDir()两个方法则用于获取指定块池目录下的rbw文件夹以及lazyPersist文件夹， 这两个方法是通过调用bpid对应的BlockPoolSlice对象的方法实现的。  
代码如下：

File getCurrentDir() {
        return currentDir;
      }
      
      protected File getRbwDir(String bpid) throws IOException {
        return getBlockPoolSlice(bpid).getRbwDir();
      }
    
      protected File getLazyPersistDir(String bpid) throws IOException {
        return getBlockPoolSlice(bpid).getLazypersistDir();
      }
    
      protected File getTmpDir(String bpid) throws IOException {
        return getBlockPoolSlice(bpid).getTmpDir();
      }

■ 数据块操作： 包括添加数据块副本、 获取数据块副本信息等操作。 这些方法底层都是在BlockPoolSlice对象上调用对应的方法  
■ 块池操作： 在当前存储目录下添加或者删除一个块池的存储， 大多是对bpSlices属性进行操作。

void addBlockPool(String bpid, Configuration c, Timer timer)
          throws IOException {
        File bpdir = new File(currentDir, bpid);
        BlockPoolSlice bp;
        if (timer == null) {
          bp = new BlockPoolSlice(bpid, this, bpdir, c, new Timer());
        } else {
          bp = new BlockPoolSlice(bpid, this, bpdir, c, timer);
        }
        bpSlices.put(bpid, bp);
      }

# 三.FsVolumeList实现 #

Datanode可以设置多个存储目录， 每一个存储目录的数据块都是由一个FsVolumeImpl对象负责管理的。 对于Datanode定义的多个存储目录， HDFS提供了接口类——FsVolumeList进行统一管理， FsVolumeList使用一个线性表保存所有的FsVolumeImpl对象， 并提供统一的接口在所有的FsVolumeImpl对象上执行操作。

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

FsVolumeList定义了两个重要的字段  
■ **Listvolumes** ：保存Datanode配置的所有FsVolumeImpl对象。 注意， 这个集合是一个只读集合， 不可以进行修改。

■ **VolumeChoosingPolicyblockChooser**： 选择一个存储目录对应的FsVolumeImpl对象来存放数据块副本。  
目前有两种策略。

> AvailableSpaceVolumeChoosingPolicy： 选择有更多可用空间的存储目录来存放副本。  
> RoundRobinVolumeChoosingPolicy： 轮询策略 \[默认\]， 轮询直到选择出第一个有足够空间的存储目录来存放副本。

FsVolumeList主要提供以下三种类型的操作。

■ **获取Datanode节点状态操作**： FsVolumeList提供了获取Datanode总容量、 剩余容量、 dfs使用量和数据块信息（ getVolumeMap()和getAllVolumesMap()） 等方法。  
■ **BlockPool相关**： FsVolumeList提供了添加和删除BlockPool的方法（ addBlockPool()、 removeBlockPool()） ， 用于在所有的FsVolumeImpl上添加和删除一个块池存储目录。

以addBlockPool()方法为例， 这个方法遍历所有的FsVolumeImpl对象， 对于每一个FsVolumeImpl对象都启动一个独立的线程级联调用该对象的addBlockPool()方法， 在该对象的存储目录下添加新的块池存储目录， 然后等待所有线程执行完毕后返回。

void addBlockPool(final String bpid, final Configuration conf) throws IOException {
        long totalStartTime = Time.monotonicNow();
        final Map<FsVolumeSpi, IOException> unhealthyDataDirs =
            new ConcurrentHashMap<FsVolumeSpi, IOException>();
        List<Thread> blockPoolAddingThreads = new ArrayList<Thread>();
        for (final FsVolumeImpl v : volumes) {
          //对于每一个FsVolumeImpl级联启动一个独立的线程
          Thread t = new Thread() {
            public void run() {
              try (FsVolumeReference ref = v.obtainReference()) {
                FsDatasetImpl.LOG.info("Scanning block pool " + bpid +
                    " on volume " + v + "...");
                long startTime = Time.monotonicNow();
    
                //调用FsVolumeImpl.addBlockPool()在对应存储目录下添加块池目录
                v.addBlockPool(bpid, conf);
                long timeTaken = Time.monotonicNow() - startTime;
                FsDatasetImpl.LOG.info("Time taken to scan block pool " + bpid +
                    " on " + v + ": " + timeTaken + "ms");
              } catch (IOException ioe) {
                FsDatasetImpl.LOG.info("Caught exception while scanning " + v +
                    ". Will throw later.", ioe);
                unhealthyDataDirs.put(v, ioe);
              }
            }
          };
          blockPoolAddingThreads.add(t);
          t.start();
        }
        //等待所有线程执行完毕
        for (Thread t : blockPoolAddingThreads) {
          try {
            t.join();
          } catch (InterruptedException ie) {
            throw new IOException(ie);
          }
        }
        long totalTimeTaken = Time.monotonicNow() - totalStartTime;
        FsDatasetImpl.LOG.info("Total time to scan all replicas for block pool " +
            bpid + ": " + totalTimeTaken + "ms");
        if (!unhealthyDataDirs.isEmpty()) {
          throw new AddBlockPoolException(unhealthyDataDirs);
        }
      }

■ 获取一个可以存储副本的存储目录： 对应于getNextVolume()方法， 这个方法的实现比较简单， 就是在blockChooser字段上调用chooseVolume()方法。 可以有两种策略， 即轮询与可用空间优先（请参考上面的字段介绍部分） 。 Datanode在存储一个新的数据块副本时， 会首先调用这个方法获取一个可以存储这个数据块副本的存储目录， 然后在这个存储目录对应的FsVolumeImpl对象上调用createRbwFile()或者createTemporaryFile()方法， 创建临时数据块文件进行写操作。

/** 
       * Get next volume.
       *
       * @param blockSize free space needed on the volume
       * @param storageType the desired {@link StorageType}
       * @param storageId the storage id which may or may not be used by
       *                  the VolumeChoosingPolicy.
       * @return next volume to store the block in.
       */
      FsVolumeReference getNextVolume(StorageType storageType, String storageId,
          long blockSize) throws IOException {
        final List<FsVolumeImpl> list = new ArrayList<>(volumes.size());
        for(FsVolumeImpl v : volumes) {
          if (v.getStorageType() == storageType) {
            list.add(v);
          }
        }
        //调用blockChooser.chooseVolume()方法获取一个可以存储副本的存储目录
        return chooseVolume(list, blockSize, storageId);
      }

> org.apache.hadoop.hdfs.server.datanode.fsdataset.RoundRobinVolumeChoosingPolicy\#chooseVolume()  
> 轮询策略会轮询所有FsVolumeImpl直到找到一个空间足够的FsVolumeImpl对象

private V chooseVolume(final int curVolumeIndex, final List<V> volumes,
                             long blockSize) throws IOException {
        // since volumes could've been removed because of the failure
        // make sure we are not out of bounds
        int curVolume = curVolumes[curVolumeIndex] < volumes.size() ? curVolumes[curVolumeIndex] : 0;
    
        int startVolume = curVolume;
        long maxAvailable = 0;
    
        while (true) {
          //返回一个有足够空间的volume
          final V volume = volumes.get(curVolume);
          
          curVolume = (curVolume + 1) % volumes.size();
          long availableVolumeSize = volume.getAvailable();
          if (availableVolumeSize > blockSize) {
            curVolumes[curVolumeIndex] = curVolume;
            return volume;
          }
    
          if (availableVolumeSize > maxAvailable) {
            maxAvailable = availableVolumeSize;
          }
    
          if (curVolume == startVolume) {
            throw new DiskOutOfSpaceException("Out of space: "
                + "The volume with the most available space (=" + maxAvailable
                + " B) is less than the block size (=" + blockSize + " B).");
          }
        }
      }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/04d45d2aba1d41129e07b613dc950fcc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/3929972f29e64aeda0c05ca8d5e44b5f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/8344b33ad69b4e238e989b49a984a8ba.png