HDFS磁盘写入数据的时候会存在磁盘写入速率不同导致磁盘使用比率差距可能会很大.但是我们可以手动移动   数据到其他的空间大的磁盘但是人为手动操作风险比较大.为了数据可以在命令行操作HDFS开发了磁盘平衡模块.该   新特性在Hadoop2.10&Hadoop3.0已经发布.本文笔者浅析磁盘均衡源码.

  @Override
  public void execute(CommandLine cmd) throws Exception {
     LOG.info("Executing \"execute plan\" command");
     Preconditions.checkState(cmd.hasOption(DiskBalancerCLI.EXECUTE));
     verifyCommandOptions(DiskBalancerCLI.EXECUTE, cmd);
     //获取执行命令的JSON文件
     String planFile = cmd.getOptionValue(DiskBalancerCLI.EXECUTE);
     Preconditions.checkArgument(planFile != null && !planFile.isEmpty(),
         "Invalid plan file specified.");
     
     String planData = null;
     //获取其中的数据
     try (FSDataInputStream plan = open(planFile)) {
       planData = IOUtils.toString(plan);
     }
     
     boolean skipDateCheck = false;
     //判断是否强制执行
     if(cmd.hasOption(DiskBalancerCLI.SKIPDATECHECK)) {
       skipDateCheck = true;
       LOG.warn("Skipping date check on this plan. This could mean we are " +
           "executing an old plan and may not be the right plan for this " +
           "data node.");
     }
     //提交执行
     submitPlan(planFile, planData, skipDateCheck);
}

   private void submitPlan(final String planFile, final String planData,
                        boolean skipDateCheck)
        throws IOException {
  Preconditions.checkNotNull(planData);
  // 解析要执行的JSON数据为执行Node执行计划
  NodePlan plan = NodePlan.parseJson(planData);
  //获取DataNode地址
  String dataNodeAddress = plan.getNodeName() + ":" + plan.getPort();
  Preconditions.checkNotNull(dataNodeAddress);
  //获取DataNode客户端协议
  ClientDatanodeProtocol dataNode = getDataNodeProxy(dataNodeAddress);
  //
  String planHash = DigestUtils.shaHex(planData);
  try {
    //真正提交磁盘平衡计划
    dataNode.submitDiskBalancerPlan(planHash, DiskBalancerCLI.PLAN_VERSION,
                                    planFile, planData, skipDateCheck);
  } catch (DiskBalancerException ex) {
    LOG.error("Submitting plan on  {} failed. Result: {}, Message: {}",
        plan.getNodeName(), ex.getResult().toString(), ex.getMessage());
    throw ex;
  }
}

  public void submitDiskBalancerPlan(String planID, long planVersion,
    String planFile, String planData, boolean skipDateCheck)
    throws IOException {
  checkSuperuserPrivilege();
  if (getStartupOption(getConf()) != StartupOption.REGULAR) {
    throw new DiskBalancerException(
        "Datanode is in special state, e.g. Upgrade/Rollback etc."
            + " Disk balancing not permitted.",
        DiskBalancerException.Result.DATANODE_STATUS_NOT_REGULAR);
  }
  //提交任务
  getDiskBalancer().submitPlan(planID, planVersion, planFile, planData,
          skipDateCheck);
}  

   public void submitPlan(String planId, long planVersion, String planFileName,
                       String planData, boolean force)
        throws DiskBalancerException {
  lock.lock();
  try {
    //检查是否已经开启平衡  
    checkDiskBalancerEnabled();
    if ((this.future != null) && (!this.future.isDone())) {
      LOG.error("Disk Balancer - Executing another plan, submitPlan failed.");
      throw new DiskBalancerException("Executing another plan",
          DiskBalancerException.Result.PLAN_ALREADY_IN_PROGRESS);
    }
    NodePlan nodePlan = verifyPlan(planId, planVersion, planData, force);
    //创建工作计划
    //实际上放到把NodePlan计划放到了ConcurrentHashMap<VolumePair, DiskBalancerWorkItem> workMap
    createWorkPlan(nodePlan);
    this.planID = planId;
    this.planFile = planFileName;
    this.currentResult = Result.PLAN_UNDER_PROGRESS;
    //执行计划
    executePlan();
  } finally {
    lock.unlock();
  }
}

  private void executePlan() {
  Preconditions.checkState(lock.isHeldByCurrentThread());
  this.blockMover.setRunnable();
  if (this.scheduler.isShutdown()) {
    this.scheduler = Executors.newSingleThreadExecutor();
  }

  this.future = scheduler.submit(new Runnable() {
    @Override
    public void run() {
      Thread.currentThread().setName("DiskBalancerThread");
      LOG.info("Executing Disk balancer plan. Plan File: {}, Plan ID: {}",
          planFile, planID);
      //根据给定的计划文件和计划ID    
      for (Map.Entry<VolumePair, DiskBalancerWorkItem> entry :
          workMap.entrySet()) {
        blockMover.setRunnable();
        //移动数据块 
        blockMover.copyBlocks(entry.getKey(), entry.getValue());
      }
    }
  });
}