Hadoop3.2.1 【 HDFS 】源码分析 : CacheManager解析

川长思鸟来 2022-11-21 11:51 275阅读 0赞

### List of articles ###

*  一. 前言
 *  二.缓存概念
 *  三. 缓存管理命令
 *  四.HDFS集中式缓存架构
 *  五.CacheManager类实现
 *  六.CacheReplicationMonitor

# 一. 前言 #

HDFS集中式缓存是由分布在Datanode上的堆外内存组成的， 并且由Namenode统一管理。

添加集中式缓存功能的HDFS集群具有以下显著的优势。  
■ 阻止了频繁使用的数据从内存中清除。  
■ 因为集中式缓存是由Namenode统一管理的， 所以HDFS客户端可以根据数据块的缓存情况调度任务， 从而提高了数据块的读性能。  
■ 数据块被Datanode缓存后， 客户端就可以使用一个新的更高效的**零拷贝机制**读取数据块。 因为数据块被缓存时已经执行了校验操作， 所以使用零拷贝读取数据块的客户端不会有读取开销。  
■ 可以提高集群的内存利用率。 当Datanode使用操作系统的buffer缓存数据块时， 对一个块重复读会导致该块的N个副本全部被送入操作系统的buffer中。 而使用集中式缓存时， 用户可以锁定这N个副本中的M个， 从而节约了N-M的内存

# 二.缓存概念 #

HDFS集中式缓存有两个主要概念。

■ 缓存指令（Cache Directive） ： 一条缓存指令定义了一个要被缓存的路径（path） 。 这些路径可以是文件夹或文件。 需要注意的是， **文件夹的缓存是非递归的**， 只有在文件夹第一层列出的文件才会被缓存。 文件夹也可以指定额外的参数， 比如缓存副本因子（replication）、 有效期等。 缓存副本因子设置了路径的缓存副本数， 如果多个缓存指令指向同一个文件，那么就用最大缓存副本因子。

缓存池（Cahce Pool） ： 缓存池是一个管理单元， 是管理缓存指令的组。 缓存池拥有类UNIX的权限， 可以限制哪个用户和组可以访问该缓存池。 缓存池也可以用于资源管理。 它可以设置一个最大限制值， 限制写入缓存池中指令的字节数。

# 三. 缓存管理命令 #

HDFS为管理员和用户提供了“hdfs cacheadmin”命令管理集中式缓存， 包括**缓存指令控制**和**缓存池控制**两个部分。  
■ **缓存指令控制**： 管理员可以调用“hdfs cacheadmin -addDirective”命令缓存指定路  
径； 调用“hdfs cacheadmin -removeDirective”命令删除指定id对应的缓存； 调用“hdfs cacheadmin -removeDirectives”命令删除指定路径的缓存； 调用“hdfs cacheadmin -listDirectives”命令显示当前所有的缓存。  
■ **缓存池控制**： 管理员可以调用“hdfs cacheadmin -addPool”命令创建一个缓存池； 调用“hdfs cacheadmin -modifyPool”命令修改一个缓存池的配置； 调用“hdfs cacheadmin -removePool”命令删除一个缓存池。

# 四.HDFS集中式缓存架构 #

用户可以通过“hdfs cacheadmin”命令或者HDFS API向Namenode发送缓存指令（Cache Directive） ， Namenode的CacheManager类会将缓存指令保存到内存的指定数据结构（CacheManager的directivesByld、 directivesByPath字段） 中， 同时在fsimage和editlog文件中记录该缓存指令。 之后Namenode的CacheReplicationMonitor类会周期性扫描命名空间和活跃的缓存指令， 以确定需要缓存或删除缓存的数据块， 并向Datanode分配缓存任务。

Namenode还负责管理集群中所有Datanode的堆外缓存， Datanode会周期性向Namenode发送缓存报告， 而Namenode会通过心跳响应向Datanode下发缓存指令（添加缓存或者删除缓存） 。

DFSClient读取数据块时会向Namenode发送ClientProtocol.getBlockLocations请求获取数据块的位置信息， Namenode除了返回数据块的位置信息外， 还会返回该数据块的缓存信息， 这样DFSClient就可以执行本地零拷贝读取缓存数据块了， 从而提高了读取效率。

# 五.CacheManager类实现 #

CacheManager类是Namenode管理集中式缓存的核心组件， 它管理着分布在HDFS集群中Datanode上的所有缓存数据块， 同时负责响应“hdfs cacheadmin”命令或者HDFS API发送的缓存管理命令

CacheManager定义了以下字段。  
■ directivesById： 以缓存指令id为key， 保存所有的缓存指令。  
■ directivesByPath： 以路径为key， 保存所有的缓存指令。  
■ cachePools： 以缓存池名称为key， 保存所有的缓存池。  
■ monitor： CacheReplicationMonitor对象， 负责扫描命名空间和活跃的缓存指令，以确定需要缓存或删除缓存的数据块， 并向Datanode分配缓存任务。

/** * Cache directives, sorted by ID. * * listCacheDirectives relies on the ordering of elements in this map * to track what has already been listed by the client. */
      private final TreeMap<Long, CacheDirective> directivesById =  new TreeMap<Long, CacheDirective>();
    
      /** * Cache directives, sorted by path */
      private final TreeMap<String, List<CacheDirective>> directivesByPath =  new TreeMap<String, List<CacheDirective>>();
    
    
    
      /** * Cache pools, sorted by name. */
      private final TreeMap<String, CachePool> cachePools =  new TreeMap<String, CachePool>();
    
    
      /** * The CacheReplicationMonitor. */
      private CacheReplicationMonitor monitor;

客户端会调用ClientProtocol.addCachePool()方法创建一个缓存池，  
NameNodeRpcServer接收了这个请求后， 会将创建缓存池请求的参数封装到一个  
CachePoolInfo对象中（包括owner、 group、 mode、 limit、 maxTtl等参数） ， 然后调用  
CacheManager.addCachePool()方法响应这个请求。 CacheManager.addCachePool()方法首先  
会验证请求参数（info变量保存） 是否合法， 然后检查CacheManager.cachePools集合中是  
否已经保存了这个缓存池。 如果CacheManager中没有这个缓存池的信息， 则调用  
CachePool.createFromInfoAndDefaults()方法根据请求参数创建一个新的CachePool对象。

createFromInfoAndDefaults()方法会将请求中没有设置的参数用默认值补齐， 然后构造CachePool对象。 成功创建CachePool对象后， addCachePool()方法会将CachePool对象放入cachePools字段中保存。 至此， 添加缓存池的操作完成。

/** * Create a cache pool. * * Only the superuser should be able to call this function. * * @param info The info for the cache pool to create. * @return Information about the cache pool we created. */
      public CachePoolInfo addCachePool(CachePoolInfo info)
          throws IOException { 
        assert namesystem.hasWriteLock();
        CachePool pool;
        try { 
    
          //验证请求的参数是否合法
          CachePoolInfo.validate(info);
    
          //获取缓存池的名字
          String poolName = info.getPoolName();
    
          // 使用TreeMap保存缓存池
          // 确认cachePools字段中不包含这个缓存池的信息
          pool = cachePools.get(poolName);
    
          //如果cachePools字段中已经有了信息， 则抛出异常
          if (pool != null) { 
            throw new InvalidRequestException("Cache pool " + poolName
                + " already exists.");
          }
          //构造CachePool对象， 对于没有设置的参数， 使用默认值补齐
          pool = CachePool.createFromInfoAndDefaults(info);
    
          //将新构造的CachePool对象放入cachePools集合中
          cachePools.put(pool.getPoolName(), pool);
        } catch (IOException e) { 
          LOG.info("addCachePool of " + info + " failed: ", e);
          throw e;
        }
        LOG.info("addCachePool of {} successful.", info);
        return pool.getInfo(true);
      }

成功地创建了缓存池之后， 客户端就会调用ClientProtocol.addCacheDirective()方法将一个路径添加到缓存中。 NameNodeRpcServer接收了这个请求后， 会将创建缓存请求的参数封装到一个CacheDirectiveInfo对象中， 然后调用CacheManager.addInternal()方法响应这个请求。  
CacheManager.addInternal()会将缓存指令对象加入directivesByld以及directivesByPath集合中保存， 然后更新缓存池的统计信息。 之后addlnternal()方法会调用setNeedsRescan()方法触发CacheReplicationMonitor执行rescan()操作。

private void addInternal(CacheDirective directive, CachePool pool) { 
        boolean addedDirective = pool.getDirectiveList().add(directive);
        assert addedDirective;
    
        //将缓存指令对象加入directivesByld集合中保存
        directivesById.put(directive.getId(), directive);
    
        //将缓存指令对象加入directivesByPath集合中保存
        String path = directive.getPath();
    
        //更新缓存池的统计信息
        List<CacheDirective> directives = directivesByPath.get(path);
        if (directives == null) { 
          directives = new ArrayList<CacheDirective>(1);
          directivesByPath.put(path, directives);
        }
        directives.add(directive);
        // Fix up pool stats
        CacheDirectiveStats stats =
            computeNeeded(directive.getPath(), directive.getReplication());
        directive.addBytesNeeded(stats.getBytesNeeded());
        directive.addFilesNeeded(directive.getFilesNeeded());
    
        //触发CacheReplicationMonitor执行rescan()操作
        setNeedsRescan();
      }

addlnternal()方法只是更新了CacheManager的directivesByld、directivesByPath以及cachePools等字段维护的缓存信息， 之后并没有对Datanode下发任何缓存指令（添加、 删除缓存） ， 而是调用setNeedsRescan()方法触发CacheReplicationMonitor执行缓存操作。

# 六.CacheReplicationMonitor #

cacheReplicationMonitor是一个线程类， 它会在Namenode启动时以及以固定的间隔扫描命名空间和活跃的缓存指令， 以确定需要缓存或删除缓存的数据块， 之后向Datanode下发缓存指令（这里的间隔是由 *dfs.namenode.path.based.cache.refresh.interval.ms* 配置项配置的， 默认为30秒）

CacheReplicationMonitor会循环调用rescan()方法执行扫描逻辑.

private void rescan() throws InterruptedException { 
        scannedDirectives = 0;
        scannedBlocks = 0;
        try { 
          namesystem.writeLock();
          try { 
            lock.lock();
            if (shutdown) { 
              throw new InterruptedException("CacheReplicationMonitor was " +
                  "shut down.");
            }
            curScanCount = completedScanCount + 1;
          } finally { 
            lock.unlock();
          }
    
          resetStatistics();
          rescanCacheDirectives();
          rescanCachedBlockMap();
          blockManager.getDatanodeManager().resetLastCachingDirectiveSentTime();
        } finally { 
          namesystem.writeUnlock();
        }
      }

参考:  
Hadoop 2.X HDFS源码剖析 – 徐鹏