hadoop，CDH如何实现 HDFS的Block数据balancer重分布

问题导读

1.如何判断一个集群是否数据倾斜？
2.balancer的原理是什么？
3.CDH集成平台如何实现数据重分布？

Hadoop的HDFS集群在使用一段时间后，各个DataNode节点的磁盘使用率肯定会出现不平衡的情况，也就是数据量层面的数据倾斜，如图：

引起这种情况的方式很多：
1.    添加新的Datanode节点
2.    人为干预将数据的副本数降低或者增加

我们都知道当HDFS出现数据不平衡的时候，就会造成MapReduce或Spark等应用程序无法很好的利用本地计算的优势，而且Datanode节点之间也没有更好的网络带宽利用率，某些Datanode节点的磁盘无法使用等等问题。

在Hadoop中，提供了hdfs balancer程序用来保证HDFS的数据平衡，我们先看一下这个程序的参数：
hdfs balancer --help
Usage: hdfs balancer
      [-policy <policy>]       the balancing policy: datanode or blockpool
      [-threshold <threshold>]    Percentage of disk capacity
      [-exclude [-f <hosts-file> | <comma-separated list of hosts>]]  Excludes the specified datanodes.
      [-include [-f <hosts-file> | <comma-separated list of hosts>]]  Includes only the specified datanodes.
      [-idleiterations <idleiterations>] Number of consecutive idle iterations (-1 for Infinite) before exit.
      [-runDuringUpgrade] Whether to run the balancer during an ongoing HDFS upgrade.This is usually not desired since it will not affect used space on over-utilized machines.

Generic options supported are
-conf <configuration file>    specify an application configuration file
-D <property=value>          use value for given property
-fs <local|namenode:port>    specify a namenode
-jt <local|resourcemanager:port> specify a ResourceManager
-files <comma separated list of files> specify comma separated files to be copied to the map reduce cluster
-libjars <comma separated list of jars> specify comma separated jar files to include in the classpath.
-archives <comma separated list of archives> specify comma separated archives to be unarchived on the compute machines.

The general command line syntax is
bin/hadoop command [genericOptions] [commandOptions]

选项的含义根据描述应该很好理解，其中-threshold参数是用来判断数据平衡的依据，值范围为0-100。默认值为10，表示HDFS达到平衡状态的磁盘使用率偏差值为10%，如果机器与机器之间磁盘使用率偏差小于10%，那么我们就认为HDFS集群已经达到了平衡的状态。

我们可以从CDH平台的CM上看到该参数是默认值和含义：

该参数具体含义为：判断集群是否平衡的目标参数，每一个 Datanode 存储使用率和集群总存储使用率的差值都应该小于这个阀值，理论上，该参数设置的越小，整个集群就越平衡，但是在线上环境中，Hadoop集群在进行balance时，还在并发的进行数据的写入和删除，所以有可能无法到达设定的平衡参数值。

参数-policy表示的平衡策略，默认为DataNode。
该参数的具体含义为：应用于重新平衡 HDFS 存储的策略。默认DataNode策略平衡了 DataNode 级别的存储。这类似于之前发行版的平衡策略。BlockPool 策略平衡了块池级别和 DataNode 级别的存储。BlockPool 策略仅适用于 Federated HDFS 服务。

参数-exclude和-include是用来选择balancer时，可以指定哪几个DataNode之间重分布，也可以从HDFS集群中排除哪几个节点不需要重分布，比如：
hdfs balancer -include CDHD,CDHA,CDHM,CDHT,CDHO

除了上面的参数会影响HDFS数据重分布，还有如下的参数也会影响重分布，
dfs.datanode.balance.bandwidthPerSec, dfs.balance.bandwidthPerSec

该默认设置：1048576(1M/s)，个人建议如果机器的网卡和交换机的带宽有限，可以适当降低该速度，一般默认就可以了。
该参数含义如下：

HDFS平衡器检测集群中使用过度或者使用不足的DataNode，并在这些DataNode之间移动数据块来保证负载均衡。如果不对平衡操作进行带宽限制，那么它会很快就会抢占所有的网络资源，不会为Mapreduce作业或者数据输入预留资源。参数dfs.balance.bandwidthPerSec定义了每个DataNode平衡操作所允许的最大使用带宽，这个值的单位是byte，这是很不直观的，因为网络带宽一般都是用bit来描述的。因此，在设置的时候，要先计算好。DataNode使用这个参数来控制网络带宽的使用，但不幸的是，这个参数在守护进程启动的时候就读入，导致管理员没办法在平衡运行时来修改这个值，如果需要调整就要重启集群。

下面简单介绍一下balancer的原理：

Rebalance程序作为一个独立的进程与NameNode进行分开执行。

步骤1：

Rebalance Server从NameNode中获取所有的DataNode情况：每一个DataNode磁盘使用情况。

步骤2：

Rebalance Server计算哪些机器需要将数据移动，哪些机器可以接受移动的数据。并且从NameNode中获取需要移动的数据分布情况。

步骤3：

Rebalance Server计算出来可以将哪一台机器的block移动到另一台机器中去。

步骤4,5,6：

需要移动block的机器将数据移动的目的机器上去，同时删除自己机器上的block数据。

步骤7：

Rebalance Server获取到本次数据移动的执行结果，并继续执行这个过程，一直没有数据可以移动或者HDFS集群以及达到了平衡的标准为止。

实战：

找一个比较空闲的的Datanode执行，建议不要在NameNode执行：

hdfs balancer -include CDHD,CDHA,CDHM,CDHT,CDHO
执行过程如下(部分)，大家可以对照上面的流程看日志，可能会更清楚一点：
16/07/11 09:35:12 INFO balancer.Balancer: namenodes = [hdfs://CDHB:8022]
16/07/11 09:35:12 INFO balancer.Balancer: parameters = Balancer.Parameters [BalancingPolicy.Node, threshold = 10.0, max idle iteration = 5, number of nodes to be excluded = 0, number of nodes to be included = 5, run during upgrade = false]
Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved
16/07/11 09:35:14 INFO net.NetworkTopology: Adding a new node: /default/192.168.1.130:50010
16/07/11 09:35:14 INFO net.NetworkTopology: Adding a new node: /default/192.168.1.131:50010
16/07/11 09:35:14 INFO net.NetworkTopology: Adding a new node: /default/192.168.1.135:50010
16/07/11 09:35:14 INFO net.NetworkTopology: Adding a new node: /default/192.168.1.138:50010
16/07/11 09:35:14 INFO net.NetworkTopology: Adding a new node: /default/192.168.1.139:50010
16/07/11 09:35:14 INFO balancer.Balancer: 2 over-utilized: [192.168.1.130:50010:DISK, 192.168.1.135:50010:DISK]
16/07/11 09:35:14 INFO balancer.Balancer: 1 underutilized: [192.168.1.131:50010:DISK]
16/07/11 09:35:14 INFO balancer.Balancer: Need to move 203.48 GB to make the cluster balanced.
16/07/11 09:35:14 INFO balancer.Balancer: Decided to move 10 GB bytes from 192.168.1.130:50010:DISK to 192.168.1.131:50010:DISK
16/07/11 09:35:14 INFO balancer.Balancer: Decided to move 10 GB bytes from 192.168.1.135:50010:DISK to 192.168.1.138:50010:DISK
16/07/11 09:35:14 INFO balancer.Balancer: Will move 20 GB in this iteration
16/07/11 09:36:00 INFO balancer.Dispatcher: Successfully moved blk_1074048042_307309 with size=134217728 from 192.168.1.130:50010:DISK to 192.168.1.131:50010:DISK through 192.168.1.130:50010
16/07/11 09:36:07 INFO balancer.Dispatcher: Successfully moved blk_1074049886_309153 with size=134217728 from 192.168.1.135:50010:DISK to 192.168.1.138:50010:DISK through 192.168.1.135:50010
16/07/11 09:36:09 INFO balancer.Dispatcher: Successfully moved blk_1074048046_307313 with size=134217728 from 192.168.1.130:50010:DISK to 192.168.1.131:50010:DISK through 192.168.1.130:50010
16/07/11 09:36:10 INFO balancer.Dispatcher: Successfully moved blk_1074049900_309167 with size=134217728 from 192.168.1.135:50010:DISK to 192.168.1.138:50010:DISK through 192.168.1.135:50010
16/07/11 09:36:16 INFO balancer.Dispatcher: Successfully moved blk_1074048061_307328 with size=134217728 from 192.168.1.130:50010:DISK to 192.168.1.131:50010:DISK through 192.168.1.130:50010
16/07/11 09:36:17 INFO balancer.Dispatcher: Successfully moved blk_1074049877_309144 with size=134217728 from 192.168.1.135:50010:DISK to 192.168.1.138:50010:DISK through 192.168.1.135:50010

如果你使用的是CDH集成平台，也可以通过CM来执行数据重分布：

步骤1：先选择HDFS组件的页面，如下：

步骤2：找到页面右侧的操作选择，从下拉框中选择数据“重新平衡”选项

步骤3：确定“重新平衡”就开始安装默认的设置规则重新分布DataNode的Block数据了，可以用CM的日志中查看具体的执行过程。