HDFS数据不均衡解决方案：基于剩余空间大小的均衡策略

前言

---

相信对于广大的Hadoop集群的使用者和维护者，集群在长时间的使用过程中，肯定或多或少碰到节点间数据不均衡的现象。比如有些节点可能磁盘使用率已经达到90%，而有些节点可能就10%。当然我们说，在使用百分比明细不均衡的情况下，我们可以用HDFS提供的Balancer工具帮我们解决这个问题。但是这不能解决所有的情况，比如说存在异构节点的集群。举一个简单的例子，集群内2个节点：A节点磁盘容量100T,B节点磁盘容量10T，如果按照默认Balancer平衡策略（按照使用百分比的策略），比如说最终会趋向于A节点使用70T，空闲30T，B节点使用7T,空闲3T。这种情况下，我们显然不希望把数据放到B节点上了，因为按照绝对值来讲，A节点剩余的30T空间显然会大很多。基于这类场景，我们可能需要一种基于剩余空间的数据均衡策略，使上面的例子最终平衡的效果变为A节点使用97T,空闲3T，B节点使用7T，空闲3T。

默认数据平衡策略的缺陷

---

在讲述本文主题之前，我们首先得理解现有策略的缺陷和不足，然后我们才能知道怎么去改进。如前言中已经提过，现有默认的Balancer策略更适应于完全同质化的节点结构（比如说相同磁盘空间），这样的话，它能够始终保持这些节点都存有差不多数据量的块数据。

但是在磁盘容量差距巨大的情况，比如说集群20个节点，10个节点拥有超大磁盘容量（100T）,而另10个节点则是普通的10T，这个时候我们当然更倾向于将更多的数据往大容量节点机器上放。一种情况，我们限制磁盘的使用率在90%，这是小磁盘容量剩余1T，这能够接受，但是大磁盘容量每台机器，就剩了10T，10个节点就是100T，可是很大的空间浪费啊。如果你想利用掉着10T,那么相对应的小磁盘容量机器会受不了，它的剩余容量绝对值已经很少了，到时机器的读写性能也可想而知。

所以针对此，笔者想到了基于剩余空间量的数据均衡策略。此策略的最终目的是使各个节点的剩余空间相等，而不是按照使用占比。

基于剩余空间大小的均衡策略

---

策略实现思路

---

这个策略目前是HDFS内部还没有实现的，所以需要我们做一点小的改动，注意这里笔者提到的是小的改动，因为里面大部分的逻辑是可以复用的。

笔者阅读了相关的代码，现有的Balaner执行逻辑大体可以分为如下：

• 1.初始化节点信息，计算集群总容量和总使用空间占比。
• 2.遍历每个节点，计算此节点目前的使用空间占比，与集群使用在空间占比做比较，计算差值。
• 3.再将步骤2计算出来的差值与平均阈值作比较，如果超出了，说明需要数据被均衡，计算公式=绝对值|差值-阈值| * 节点空间，这就是需要被平衡的数据量。
• 4.再将此值与最大可移动数据量做比较，取2者较小值。

到了这里，主要逻辑就完成了，后续就是Balancer程序将节点分配好。这里用数字来解释一下，比如说一个集群（假设含2个节点）使用率目前20%，A使用5%，B使用35%，然后算出A节点的差值为20%-5%=%15，与平衡阈值参数（假设1%）比较，发现相差15%-1%=14%，也就是需要平衡14%的数据，然后算出移动数据，最终的平衡效果将会是A（5+14）%=19%，B就是（35-14）%=21%。

OK，回到正题，如果我们想实现一套新的基于物理剩余空间的平衡策略，我们只需要做如下细微的改动：

• 总节点使用占比应换为节点平均剩余空间大小。
• 节点的差值=(节点当前剩余空间-节点平均剩余空间)/节点平均剩余空间(因为这个地方我们需要使用物理大小做计算，所以百分比差值得自己计算)
• 如果需要移动数据，可移动空间=节点当前剩余空间-节点平均剩余空间。

后面差值与阈值的比较同理，这里就不介绍了。在这种算法下，最后节点内的剩余可用空间会趋向于相同大小的情况，也就是我们所希望看到的基于剩余空间大小的均衡。

策略代码实现

---

下面到了大家最为关系的代码实现部分了。笔者也为大家做了实现。

目前Balancer Policy策略定义还算比较灵活，为用户自定义策略开放了一些接口，也方便了笔者的实现，首先定义一个新的策略类，如下：

  /**
   * 基于剩余空间大小的策略类
   */
  static class NodeRemaing extends BalancingPolicy {
    static final Pool INSTANCE = new Pool();
    private NodeRemaing() {}

    @Override
    String getName() {
      return "nodeRemaing";
    }

    @Override
    void accumulateSpaces(DatanodeStorageReport r) {
      for(StorageReport s : r.getStorageReports()) {
        final StorageType t = s.getStorage().getStorageType();
        totalCapacities.add(t, s.getCapacity());
        // 按照使用剩余空间标准来算
        totalUsedSpaces.add(t, s.getRemaining());
      }
    }

    @Override
    Double getUtilization(DatanodeStorageReport r, StorageType t) {
      long capacity = 0L;
      long remgainUsed = 0L;
      for (StorageReport s : r.getStorageReports()) {
        if (s.getStorage().getStorageType() == t) {
          capacity += s.getCapacity();
          remgainUsed += s.getRemaining();
        }
      }

      // 同样返回的使用占比为剩余物理空间大小
      return capacity == 0L ? Double.valueOf(0) : Double.valueOf(remgainUsed);
    }

    @Override
    double getAvgUtilization(StorageType t) {
      // 计算节点平均剩余空间大小
      return avgUtilizations.get(t) * totalCapacities.get(t);
    }
  }

将这新的策略类加到解析方法里去，这样才能被激活使用：

  /** Get all {@link BalancingPolicy} instances*/
  // 加入新的策略类
  static BalancingPolicy parse(String s) {
    final BalancingPolicy [] all = {BalancingPolicy.Node.INSTANCE,
                                    BalancingPolicy.Pool.INSTANCE,
                                    BalancingPolicy.NodeRemaing.INSTANCE};
    for(BalancingPolicy p : all) {
      if (p.getName().equalsIgnoreCase(s))
        return p;
    }
    throw new IllegalArgumentException("Cannot parse string "" + s + """);
  }

核心算法逻辑的改动在方法Balancer.init(List reports)，改动如下：

  private long init(List<DatanodeStorageReport> reports) {
    ...

        double utilizationDiff;
        double thresholdDiff;
        long maxSize2Move;
        long capacity = getCapacity(r, t);
        // TODO:
        // 目前基于剩余空间策略类不太兼容现有逻辑，需要做一下判断，其实更好的做法是再定义一个接口
        if (policy instanceof BalancingPolicy.NodeRemaing) {
          // 算出百分比差值，utilization当前节点的剩余空间，average：集群平均剩余空间
          utilizationDiff = (utilization - average) / average;
          // 比较与阈值的差值
          thresholdDiff = Math.abs(utilizationDiff) - threshold;

          // 算出需要移动的空间大小
          long moveSize = (long) Math.abs(utilization - average);
          // 与最大可移动空间大小比较，不允许大于此值
          maxSize2Move = Math.max(maxSizeToMove, moveSize);
        } else {
          // 原策略逻辑维持不动
          utilizationDiff = utilization - average;
          thresholdDiff = Math.abs(utilizationDiff) - threshold;
          maxSize2Move = computeMaxSize2Move(capacity, getRemaining(r, t),
              utilizationDiff, maxSizeToMove);
        }
        ...
}

要做的改动不多，就是上面提到的这些，关键点在于是否把这个问题想清楚了，想明白了，解决起来就不是什么难事了。目前这个改动笔者还没来得及测试，感兴趣的朋友们可以自己测测效果。代码已经上传到我的github上了，链接：https://github.com/linyiqun/open-source-patch/tree/master/hdfs/others/HDFS-RemaingBasedBalancerPolicy

其实按照剩余空间大小均衡的策略在同构集群内其实是等同于按照使用占比平衡的策略，因为每个节点的容量是等同的，只是从反面来算罢了，大家可以好好理解一下。