首页 > 其他分享> > HDFS多rack分布的block placement policy设计实现

HDFS多rack分布的block placement policy设计实现

2021-07-03 13:32:08 作者：互联网

文章目录

前言
HDFS多rack分布的block placement policy
- 多rack分布的policy实现思路
旧block placement的到新block placement的迁移

前言

众所周知，HDFS拥有3副本来保证其数据的高可用性。而且HDFS对着三个副本的位置放置也是有专心设计的，2个副本放在同一个rack（不同节点），另外一个副本放在另外的一个rack上。在这样的放置策略下，这个副本数据能容忍一个节点的crash甚至是一个rack机器的crash。但这里所提及的"rack“的概念是集群admin给HDFS定义的rack的概念，它是一个逻辑上的概念。它可以简单的是一个物理rack，也可以是一组rack的集合。不过它们一个共同的特征是rack与rack之间是能够隔离开的。HDFS目前默认的block放置策略在理论上是能够容忍一个rack的掉线，但是在实际大规模集群的运行过程中，默认的放置策略还是不能够完全保证数据高可用性的要求。举例来说，笔者最近在生产集群上频繁碰到因为数据3副本同时不可用导致的用户missing block问题。后来发现是因为集群在进行按rack的rolling升级，每次会有长达1小时的rack shutdown的时间。然后在此期间，偶发的其它rack的机器的dead，就会造成这种零星的missing block情况。这个问题本质上的原因是因为掉了1个rack，导致存在于此rack上的2副本无法使用，进而导致了大概率数据无法使用的情况。针对此情况，我们尝试对现有的block placement策略进行改造，来解决这个问题。

HDFS多rack分布的block placement policy

我们知道，HDFS的block位置存放一方面是基于内部的block placement policy实现，还有另一方面是admin给定的topology设置。Topology是这里的作用是规定了什么节点属于哪些rack。

如果我们不想对现有HDFS placement policy进行改造来解决上面提到的问题，改变集群使用的topology也是能够奏效的。让topology里的rack映射到实际更多的物理rack能一定程度上减缓这个问题。但这里笔者说的是减缓，不是彻底解决。因为topology里的rack的scope尽管变大了，但是还是存在俩副本落在一个物理rack上的情况的。

因此，从根本上解决block按照rack分布的问题的办法就是改变其block placement的policy策略来做。目标期望的block placement分布也很简单，就是3副本同时分布在3个rack上。这样的话，集群就能够极大地容忍一个rack掉线导致的数据不可用问题。

多rack分布的policy实现思路

鉴于HDFS默认的block placement policy实现逻辑比较复杂，我们不倾向于直接改它里面的过程逻辑。而是新增一个新的policy类，然后覆盖其部分方法，以此来达到我们按照rack分布的block placement policy的实现。

这里会涉及到的几个关键的方法：

chooseLocalRack
chooseRemoteRack
isGoodDatanode
verifyBlockPlacement
isMovable

以上5个方法和HDFS的block placement选择息息相关。上面方法的前2个方法是首先需要覆盖的方法，因为在多rack分布策略下，chooseLocalRack的语义此时将转变为和chooseRemoteRack一样的语义。chooseRemoteRack方法其实和default policy的实现一致，以防后续这个方法也会需要变动，笔者暂时对chooseRemoteRack方法进行了复制，代码如下：

public class BlockPlacementPolicyWithRackSeparated extends BlockPlacementPolicyDefault {

  private static ThreadLocal<Integer> CHOOSE_TARGET_FAILED_COUNT =
      new ThreadLocal<Integer>() {
    @Override
    protected Integer initialValue() {
      return 0;
    }
  };

  @Override
  protected DatanodeStorageInfo chooseLocalRack(Node localMachine, Set<Node> excludedNodes,
      long blocksize, int maxNodesPerRack, List<DatanodeStorageInfo> results,
      boolean avoidStaleNodes, EnumMap<StorageType, Integer> storageTypes) throws NotEnoughReplicasException {

    // no local machine, so choose a random machine
    if (localMachine == null) {
      return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
          maxNodesPerRack, results, avoidStaleNodes, storageTypes);
    }
    final String localRack = localMachine.getNetworkLocation();

    try {
      // override default chooseLocalRack method and choose one from the remote rack
      return chooseRandom("~" + localRack, excludedNodes,
          blocksize, maxNodesPerRack, results, avoidStaleNodes, storageTypes);
    } catch (NotEnoughReplicasException e) {
        return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
            maxNodesPerRack, results, avoidStaleNodes, storageTypes);
      }
  }

  @Override
  protected void chooseRemoteRack(int numOfReplicas, DatanodeDescriptor localMachine,
    Set<Node> excludedNodes, long blocksize, int maxReplicasPerRack,
    List<DatanodeStorageInfo> results, boolean avoidStaleNodes, EnumMap<StorageType,
    Integer> storageTypes) throws NotEnoughReplicasException {
    int oldNumOfReplicas = results.size();
    // randomly choose one node from remote racks
    try {
      chooseRandom(numOfReplicas, "~" + localMachine.getNetworkLocation(),
      excludedNodes, blocksize, maxReplicasPerRack, results,
      avoidStaleNodes, storageTypes);
      } catch (NotEnoughReplicasException e) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Failed to choose remote rack (location = ~"
              + localMachine.getNetworkLocation() + "), fallback to local rack", e);
        }
        chooseRandom(numOfReplicas-(results.size()-oldNumOfReplicas),
            localMachine.getNetworkLocation(), excludedNodes, blocksize,
            maxReplicasPerRack, results, avoidStaleNodes, storageTypes);
      }
  }
  ...
}

其次紧跟着的isGoodDatanode方法，这个方法是用在chooseRemoteRack里选好一个target后判断此candidate是否是一个合格的datanode的检查方法，如果不合格，则会进行马上的下一次选择。在新的policy实现中，我们主要对rack的值进行判断，确保每个副本的location所属的rack都是独立的。

...
  @Override
  boolean isGoodDatanode(DatanodeDescriptor node, int maxTargetPerRack,
      boolean considerLoad, List<DatanodeStorageInfo> results, boolean avoidStaleNodes) {
    if (!super.isGoodDatanode(node, maxTargetPerRack, considerLoad, results, avoidStaleNodes)) {
      return false;
    }

    // the chosen node rack
    String rack = node.getNetworkLocation();
    Set<String> rackNames = new HashSet<>();
    rackNames.add(rack);

    for (DatanodeStorageInfo info : results) {
      rack = info.getDatanodeDescriptor().getNetworkLocation();
      if (!rackNames.contains(rack)) {
        rackNames.add(rack);
      } else {
        LOG.warn("Chosen node failed since there is nodes in the same rack.");
        return false;
      }
    }

    return true;
  }
...

最后是verifyBlockPlacement和isMovable方法，前者用于判断现有block副本的placement是否满足多rack分布模式，后者方法是用在Balancer里判断移动某个location的block是否是可移动的，如果break了block placement policy，则是不可移的。方法如下：

...
  @Override
  public BlockPlacementStatus verifyBlockPlacement(DatanodeInfo[] locs,
      int numberOfReplicas) {
    if (locs == null) {
      locs = DatanodeDescriptor.EMPTY_ARRAY;
    }

    if (!clusterMap.hasClusterEverBeenMultiRack()) {
      // only one rack
      return new BlockPlacementStatusDefault(1, 1);
    }

    int minRacks = 3;
    minRacks = Math.min(minRacks, numberOfReplicas);
    // 1. Check that all locations are different.
    // 2. Count locations on different racks.
    Set<String> racks = new TreeSet<>();
    for (DatanodeInfo dn : locs) {
      racks.add(dn.getNetworkLocation());
    }
    return new BlockPlacementStatusDefault(racks.size(), minRacks);
  }

  @Override
  public boolean isMovable(Collection<DatanodeInfo> locs, DatanodeInfo source,
      DatanodeInfo target) {
    Set<String> rackNames = new HashSet<>();
    for (DatanodeInfo dn : locs) {
      rackNames.add(dn.getNetworkLocation());
    }

    rackNames.remove(source.getNetworkLocation());
    rackNames.add(target.getNetworkLocation());

    return rackNames.size() >= locs.size() ? true : false;
  }
...

以上就是实现多rack分布的新block placement policy的主要方法实现。笔者已经在ut测试方法中验证过此逻辑的正确性了，但还未在生产集群部署过此policy，目前仅供大家参考实现。

旧block placement的到新block placement的迁移

上面实现完多rack分布的placement policy就意味着已经完美解决了上面提到的问题了呢？答案是没有。有些事情没有像想象的那么简单。多rack的placement policy可以解决的是新数据写入的block分布的问题。但是对于集群里大量的原有placement的block，我们还是需要去做里面的placement的转化的。如何能够平滑，透明地迁移老placement的block数据也是一个难点问题。

对于block placement的迁移，笔者调研了现有可用的一些方法，可归纳为大致两类：

基于Server端的迁移方案。大意是说让NN端帮助我们做这个事情，但在NN端的优点是方便，但是隐患比较大，第一NN端做，NN做block placement的切换转换速度不好控制，会增加NN的load压力，然后影响到NN的正常处理性能。目前知道的NN server端的方法是依赖其服务启动后的replicationQueuesInitializer线程做block placement的扫描检查。如果检查出有block违背了placement的policy，则会触发后续的block placement的重复制分布。
基于Client端的迁移方案。Client端的方案相较于Server端的，迁移速度更易于控制，而且影响小。这里面有2种可选的方案：
1）使用现有Balancer工具做block的迁移。Balancer目前在搬运block时会遵循policy的规则设置(相关JIRA HDFS-9007)。因此我们可以利用它这一点来帮助我们做迁移。但是缺点在于此方法效率会遍低，因为Balancer只做数据使用量不均衡节点数据的搬迁。假设集群数据本身已经十分均衡了，Balancer能够搬运的数据就会比较有限了，迁移的预期效果也不会如我们所预期的那么好了。
2）基于Fsck path级别的block迁移。Balancer工具是基于block做的迁移，其实我们还可以支持按照path(目录或文件)级别的block迁移。每次指定一个path，然后拿到这些path下所属的文件block进行迁移。迁移完一个path后，再进行下一个path的迁移。每次path的路径根据其实际文件数的规模，进行调整即可。这样的话，整个迁移过程就会比较平滑，而且影响较小。不过目前fsck对path级别的block迁移功能的支持在3.3版本上才有（相关JIRA HDFS-14053），需要我们进行patch的一个backport。

OK，以上就是本文所阐述的关于引入多rack分布的新block placement policy实现来提高数据可用性的主要内容了，其间也提到了关于如果做老数据的block placement迁移的问题。希望对大家有所收获。

标签：HDFS,placement,policy,迁移,rack,block
来源： https://blog.csdn.net/Androidlushangderen/article/details/118437716