hadoop源码分析系列（五）——org.apache.hadoop.hdfs包之balancer篇

gtdfe

摘要: 首先说明下均衡器相关的原理知识：hadoop默认的复本布局策略是在发起请求的客户端存放一个复本，如果这个客户端在集群以外，那就选择一个不是太忙，存储不是太满的节点来存放，第二个复本放在与第一个复本相同的机 ...

首先说明下均衡器相关的原理知识：
hadoop默认的复本布局策略是在发起请求的客户端存放一个复本，如果这个客户端在集群以外，那就选择一个不是太忙，存储不是太满的节点来存放，第二个复本放在与第一个复本相同的机架但是不同节点上，第三个放在与第二个和第一个复本不同的机架上，原则是尽量避免在相同的机架上放太多的复本。
为了解决部分datanode过于繁忙的问题，hadoop提供了一个守护进程来重新布局数据块，按照默认的规则来重新分布，可以通过start-balancer.sh来启动均衡器
hadoop提供了两个队均衡器的配置参数，分别是dfs.balance.bandwidthPerSec和threshold ，前者限制复制数据的带宽，后者规定了集群均衡的阈值

下面是对Balancer类的分析：
首先看下主要的属性:
MAX_NUM_CONCURRENT_MOVES:允许同时并发复制的块数 默认为5
threshold：阈值 默认10%
支持的协议类型：NamenodeProtocol ClientProtocol
四个链表：
overUtilizedDatanodes：过载的datanode信息
aboveAvgUtilizedDatanodes：大于阈值的datanode信息
belowAvgUtilizedDatanodes：小于阈值的datanode信息
underUtilizedDatanodes：空载的datanode信息
连个集合：
sources：源地址集合
targets：目标地址集合
连个Map：
globalBlockList：记录balance过程中全部块和balance的块的对应信息
datanodes：记录datanode和balance的块的对应信息

两个线程池
moverExecutor：用于移动的线程池 默认1000
dispatcherExecutor：用于分发的线程池 默认200

首先看主要的概念类：
1、BalancerBlock类来跟踪Balancer过程中的块信息，其中包括块信息和所对应的datanode的列表信息
2、BytesMoved类记录移动的字节数
3、MovedBlocks类，这个类维护了两个窗口，一个是旧数据，一个是最近的数据，之所以说最近是为了区别最新，要保证窗口中存储1.5小时内被移动的块的信息，可以通过参数dfs.balancer.movedWinWidth来配置窗口时间，过期的块信息会被删除。
4、NodeTask类代表了一个需要复制byte的对象，这个对象存在于source节点，包含了目标节点和byte长度

5、BalancerDatanode类主要是在balance过程中跟踪datanode的信息，其中记录了datanode中多个PendingBlockMove
初始化的时候设置了节点的磁盘利用率、最大可以移动的块大小等等。

6、Source类继承了BalancerDatanode，继承了BalancerDatanode类中记录的节点信息，又新定义了几个特殊的针对源datanode的操作：
dispatchBlocks():分发线程的主方法，首先选择要移动的块，然后调用给代理源发送移动请求，当源节点的利用率小于阈值的时候，向请求namenode请求更多的块，当分发了足够的块或是接收到了namenode发出的足够块信息或是超过运行的限制时间时停止（默认20分钟）。
chooseNextBlockToMove()：返回一个可以被立即分发的块的信息

7、内部类PendingBlockMove记录了块移动过程中的跟踪信息
包括要移动的块、源地址、目标地址、代理源，这里有必要说下代理源的原理，源node先把数据块拷贝给代理，然后由代理复制到目标节点，这么设计的好处我想应该是为了保证数据不丢失吧，但是也相当于多了一个中间环节，可以说有利有弊吧。
这个内部类提供了下面的方法：
chooseBlockAndProxy()：为本次拷贝选择要拷贝的块和代理
markMovedIfGoodBlock(BalancerBlock block)：如果块是可以移动的，那么标记并放到movedBlocks队列中
这里面所说的可以移动是指能马上移动，并且已经找到了一个不繁忙的代理
dispatch（）：把块分发给代理
sendRequest(DataOutputStream out)：发出replace命令
receiveResponse(DataInputStream in)：解析操作结果


下面分析Balancer类的主方法：

createNamenode(Configuration conf)：和namenode创建一个基于NamenodeProtocol协议的连接
getUtilization(DatanodeInfo datanode)：获得datanode的使用情况
checkAndMarkRunningBalancer()：用标记文件的方法保证守护进程的唯一性
chooseSource(BalancerDatanode target,Iterator<Source> sourceCandidates, boolean onRackSource) 选择源节点，最后一个参数指出源和目标是不是同一rack，选择好更新源队列和目标队列
chooseTarget(Source source,Iterator<BalancerDatanode> targetCandidates, boolean onRackTarget)和选择源类似
dispatchBlockMoves()：提交分发请求到分发线程池中，并等待复制结束
waitForMoveCompletion()：通过检查target方法的pendingMove队列判断复制是否结束。

总结下balance的步骤：
1、从namenode获取datanode磁盘使用情况
2、计算哪些节点需要把哪些数据移动到哪里
3、分别移动，完成后删除旧的block信息
4、循环执行，直到达到平衡标准