Hadoop-balancer执行原理

核心类在

org.apache.hadoop.hdfs.server.balancer.Balancer

均衡算法 伪代码

while(true) {1.获取需要迁移的字节数if(需要迁移字节数 == 0) {return "成功，无需迁移";}2.选择需要迁移的节点if(需要移动的数据 == 0) {return "没有需要移动的块"}3.开始并行迁移4.清空列表5.Thread.sleep(2*conf.getLong("dfs.heartbeat.interval", 3));
}

获取所有的data node节点，计算

initNodes(client.getDatanodeReport(DatanodeReportType.LIVE));

initNodes()函数如下:

计算平均使用量long totalCapacity=0L, totalUsedSpace=0L;for (DatanodeInfo datanode : datanodes) {if (datanode.isDecommissioned() || datanode.isDecommissionInProgress()) {continue; // ignore decommissioning or decommissioned nodes}totalCapacity += datanode.getCapacity();totalUsedSpace += datanode.getDfsUsed();}

当前集群的平均使用率(是当前使用的空间/总空间*100)，注意这个是百分比计算后再乘100的值，不是百分比

this.avgUtilization = ((double)totalUsedSpace)/totalCapacity*100;

四个队列

1.aboveAvgUtilizedDatanodes(超过集群平均使用率 && 低于集群平均使用率+阀值)

2.overUtilizedDatanodes(超过集群平均使用率+阀值)

3.belowAvgUtilizedDatanodes(低于集群平均使用率 && 超过集群平均使用率-阀值)

4.underUtilizedDatanodes(低于集群平均使用率-阀值)

2个参数

overLoadedBytes 超过负载值的字节

underLoadedBytes低于负载值的字节

//注意这里的阈值默认是10D，这里不是百分比计算集群平均使用率如果为0.5不是50%，而相当于0.5%
//所以如果是0.5-10D就变成负数了，一般来说肯定是小于当前节点使用率的，除非当前节点使用率特别大
//比如当前节点使用率为20，则用百分比来说就是使用了20%，这肯定就超于阈值了，于是这个节点的数据
//就需要均衡了
for (DatanodeInfo datanode : datanodes) {if(当前节点使用率 > 集群平均使用率) {if(当前节点使用率 <=(集群平均使用率+阀值) && 当前节点使用率 > 集群平均使用率) {创建一个BalancerDatanodeaboveAvgUtilizedDatanodes.save(当前节点)}else {overUtilizedDatanodes.save(当前节点)overLoadedBytes += (当前节点使用率-集群平均使用率-阀值)*当前节点总数据量/100}}else {创建一个BalancerDatanodeif(当前节点使用率>=(集群平均使用率-阀值) && 当前节点使用率<集群平均使用率) {belowAvgUtilizedDatanodes.save(当前节点)}else {underUtilizedDatanodes.save(当前节点)underLoadedBytes += (集群平均使用率-阀值-当前节点使用率)*当前节点总数据量/100}}
}均衡器只会执行 overUtilizedDatanodes 和 underUtilizedDatanodes队列中的集群

BalancerDatanode()构造函数

if(当前节点使用率 >= 集群平均使用率+阀值 || 当前节点使用率 <= 集群平均使用率-阀值) {一次移动的数据量 = 阀值*当前节点总容量/100
}
else {一次移动的数据量 = (集群平均使用率-当前节点使用率) * 当前节点总容量/100
}
一次移动的数据量 = min(当前节点剩余使用量,一次移动的数据量)
一次移动的数据量 = (一次移动数据量上限10G，一次移动的数据量)

chooseNodes()函数

chooseNodes(true);	 //首先在相同机架中迁移
chooseNodes(false);	 //在不同机架中迁移chooseNodes(boolean onRack) {chooseTargets(underUtilizedDatanodes.iterator(), onRack);chooseTargets(belowAvgUtilizedDatanodes.iterator(), onRack);chooseSources(aboveAvgUtilizedDatanodes.iterator(), onRack);
}chooseTargets() {for(源节点 source : overUtilizedDatanodes列表) {选择目标节点(source)}
}选择目标节点(source) {while() {1.从候选队列中找到一个节点2.如果这个可转移的数据已经满了continue3.if(在相同机架中转移)4.if(在不同机架中转移)5.创建NodeTask}
}//和chooseTargets函数类似
chooseSources() {for(目标节点 target : underUtilizedDatanodes) {选择源节点()}
}选择源节点(target) {while() {1.从候选队列中找到一个节点2.如果这个节点可转移的数据已经满了continue3.if(在相同机架中转移)4.if(在不同机架中转移)5.创建NodeTask}
}控制台或者日志上会显示  Decided to move 3.55 GB bytes from source_host:50010 to target_host:50010

开始并行迁移数据

    for (Source source : sources) {futures[i++] = dispatcherExecutor.submit(source.new ());}

BlockMoveDispatcher线程

1.选择要迁移的节点 chooseNextBlockToMove()
2.if(要迁移的节点 != null) {//启动数据迁移，创建一个新线程发送接收数据scheduleBlockMove()}
3.获取block列表，继续下一轮迁移

发送和接收数据块的dispatch()函数

//使用阻塞IO的方式发送数据并接收返回的结果sock.connect(NetUtils.createSocketAddr(target.datanode.getName()), HdfsConstants.READ_TIMEOUT);sock.setKeepAlive(true);out = new DataOutputStream( new BufferedOutputStream(sock.getOutputStream(), FSConstants.BUFFER_SIZE));sendRequest(out);in = new DataInputStream( new BufferedInputStream(sock.getInputStream(), FSConstants.BUFFER_SIZE));receiveResponse(in);bytesMoved.inc(block.getNumBytes());