核心类在
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(当前节点使用率 <=(集群平均使用率+阀值) && 当前节点使用率 > 集群平均使用率) { 创建一个BalancerDatanode aboveAvgUtilizedDatanodes.save(当前节点) } else { overUtilizedDatanodes.save(当前节点) overLoadedBytes += (当前节点使用率-集群平均使用率-阀值)*当前节点总数据量/100 } } else { 创建一个BalancerDatanode if(当前节点使用率>=(集群平均使用率-阀值) && 当前节点使用率<集群平均使用率) { 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.如果这个可转移的数据已经满了continue 3.if(在相同机架中转移) 4.if(在不同机架中转移) 5.创建NodeTask } } //和chooseTargets函数类似 chooseSources() { for(目标节点 target : underUtilizedDatanodes) { 选择源节点() } } 选择源节点(target) { while() { 1.从候选队列中找到一个节点 2.如果这个节点可转移的数据已经满了continue 3.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());
相关推荐
赠送jar包:hadoop-mapreduce-client-jobclient-2.6.5.jar; 赠送原API文档:hadoop-mapreduce-client-jobclient-2.6.5-javadoc.jar; 赠送源代码:hadoop-mapreduce-client-jobclient-2.6.5-sources.jar; 赠送...
赠送jar包:hadoop-yarn-client-2.6.5.jar; 赠送原API文档:hadoop-yarn-client-2.6.5-javadoc.jar; 赠送源代码:hadoop-yarn-client-2.6.5-sources.jar; 赠送Maven依赖信息文件:hadoop-yarn-client-2.6.5.pom;...
hadoop-annotations-3.1.1.jar hadoop-common-3.1.1.jar hadoop-mapreduce-client-core-3.1.1.jar hadoop-yarn-api-3.1.1.jar hadoop-auth-3.1.1.jar hadoop-hdfs-3.1.1.jar hadoop-mapreduce-client-hs-3.1.1.jar ...
赠送jar包:hadoop-yarn-common-2.6.5.jar 赠送原API文档:hadoop-yarn-common-2.6.5-javadoc.jar 赠送源代码:hadoop-yarn-common-2.6.5-sources.jar 包含翻译后的API文档:hadoop-yarn-common-2.6.5-javadoc-...
hadoop-eclipse-plugin-2.7.3和2.7.7的jar包 hadoop-eclipse-plugin-2.7.3和2.7.7的jar包 hadoop-eclipse-plugin-2.7.3和2.7.7的jar包 hadoop-eclipse-plugin-2.7.3和2.7.7的jar包
hadoop-eclipse-plugin-1.2.1hadoop-eclipse-plugin-1.2.1hadoop-eclipse-plugin-1.2.1hadoop-eclipse-plugin-1.2.1
该资源包里面包含eclipse上的hadoop-1.2.1版本插件的jar包和hadoop-1.2.1.tar.gz,亲测可用~~请在下载完该包后解压,将hadoop-1.2.1放置于Eclipse\plugins目录下,然后重启eclipse,将hadoop-1.2.1.tar.gz放到D:\...
赠送jar包:hadoop-common-2.7.3.jar; 赠送原API文档:hadoop-common-2.7.3-javadoc.jar; 赠送源代码:hadoop-common-2.7.3-sources.jar; 赠送Maven依赖信息文件:hadoop-common-2.7.3.pom; 包含翻译后的API文档...
hadoop-common-2.4.1.jar,是学习基础的Hadoop必须的包
hadoop-eclipse-plugin-2.7.4.jar和hadoop-eclipse-plugin-2.7.3.jar还有hadoop-eclipse-plugin-2.6.0.jar的插件都在这打包了,都可以用。
flink-shaded-hadoop-3下载
Apache Hadoop (hadoop-3.3.4.tar.gz)项目为可靠、可扩展的分布式计算开发开源软件。官网下载速度非常缓慢,因此将hadoop-3.3.4 版本放在这里,欢迎大家来下载使用! Hadoop 架构是一个开源的、基于 Java 的编程...
hadoop-eclipse-plugin-3.1.1, hadoop eclipse 插件 3.1.1
找不到与hadoop-2.9.2版本对应的插件,手动生成的hadoop-eclipse-plugin-2.9.2版本,
赠送jar包:hadoop-yarn-server-resourcemanager-2.6.0.jar; 赠送原API文档:hadoop-yarn-server-resourcemanager-2.6.0-javadoc.jar; 赠送源代码:hadoop-yarn-server-resourcemanager-2.6.0-sources.jar; 赠送...
赠送jar包:hadoop-mapreduce-client-common-2.6.5.jar; 赠送原API文档:hadoop-mapreduce-client-common-2.6.5-javadoc.jar; 赠送源代码:hadoop-mapreduce-client-common-2.6.5-sources.jar; 赠送Maven依赖信息...
hadoop2 lzo 文件 ,编译好的64位 hadoop-lzo-0.4.20.jar 文件 ,在mac 系统下编译的,用法:解压后把hadoop-lzo-0.4.20.jar 放到你的hadoop 安装路径下的lib 下,把里面lib/Mac_OS_X-x86_64-64 下的所有文件 拷到 ...
1.安装 Hadoop-gpl-compression 1.1 wget http://hadoop-gpl-compression.apache-extras.org.codespot.com/files/hadoop-gpl-compression-0.1.0-rc0.tar.gz 1.2 mv hadoop-gpl-compression-0.1.0/lib/native/Linux-...
hadoop-2.6.0版本的apache和cdh版本
赠送jar包:hadoop-auth-2.5.1.jar; 赠送原API文档:hadoop-auth-2.5.1-javadoc.jar; 赠送源代码:hadoop-auth-2.5.1-sources.jar; 赠送Maven依赖信息文件:hadoop-auth-2.5.1.pom; 包含翻译后的API文档:hadoop...