hbase源码系列（一）Balancer 负载均衡
看源碼很久了，終于開始動手寫博客了，為什么是先寫負載均衡呢，因為一個室友入職新公司了，然后他們遇到這方面的問題，某些機器的硬盤使用明顯比別的機器要多，每次用hadoop做完負載均衡，很快又變回來了。
首先我們先看HMaster當中怎么初始化Balancer的，把集群的狀態穿進去，設置master，然后執行初始化。
//initialize load balancer this.balancer.setClusterStatus(getClusterStatus()); this.balancer.setMasterServices(this); this.balancer.initialize();然后調用是在HMaster的balance()方法當中調用
Map<TableName, Map<ServerName, List<HRegionInfo>>> assignmentsByTable =this.assignmentManager.getRegionStates().getAssignmentsByTable();List<RegionPlan> plans = new ArrayList<RegionPlan>(); //Give the balancer the current cluster state. this.balancer.setClusterStatus(getClusterStatus()); //針對表來做平衡，返回平衡方案，針對全局，可能不是最優解 for (Map<ServerName, List<HRegionInfo>> assignments : assignmentsByTable.values()) {List<RegionPlan> partialPlans = this.balancer.balanceCluster(assignments);if (partialPlans != null) plans.addAll(partialPlans); }可以看到它首先獲取了當前的集群的分配情況，這個分配情況是根據表的 Map<TableName, Map<ServerName, List<HRegionInfo>>，然后遍歷這個map的values，調用balancer.balanceCluster(assignments) 來生成一個partialPlans，生成RegionPlan(Region的移動計劃) 。
我們就可以切換到StochasticLoadBalancer當中了，這個是默認Balancer具體的實現了，也是最好的實現，下面就說說這玩意兒咋實現的。
看一下注釋，這個玩意兒吹得神乎其神的，它說它考慮到了這么多因素：
* <ul>
* <li>Region Load</li> Region的負載
* <li>Table Load</li> ?表的負載
* <li>Data Locality</li> 數據本地性
* <li>Memstore Sizes</li> 內存Memstore的大小
* <li>Storefile Sizes</li> 硬盤存儲文件的大小
* </ul>
好，我們從balanceCluster開始看吧，一進來第一件事就是判斷是否需要平衡
//不需要平衡就退出 if (!needsBalance(new ClusterLoadState(clusterState))) {return null; }平衡的條件是：負載最大值和最小值要在平均值（region數/server數）的+-slop值之間, 但是這個平均值是基于表的，因為我們傳進去的參數clusterState就是基于表的。
// Check if we even need to do any load balancing // HBASE-3681 check sloppiness first float average = cs.getLoadAverage(); // for logging //集群的負載最大值和最小值要在平均值的+-slop值之間 int floor = (int) Math.floor(average * (1 - slop)); int ceiling = (int) Math.ceil(average * (1 + slop)); if (!(cs.getMinLoad() > ceiling || cs.getMaxLoad() < floor)) {.....return false; } return true;?
? 如果需要平衡的話，就開始計算開銷了
// Keep track of servers to iterate through them. Cluster cluster = new Cluster(clusterState, loads, regionFinder); //計算出來當前的開銷 double currentCost = computeCost(cluster, Double.MAX_VALUE); double initCost = currentCost; double newCost = currentCost; for (step = 0; step < computedMaxSteps; step++) {
//隨機挑選一個"選號器"
int pickerIdx = RANDOM.nextInt(pickers.length);
RegionPicker p = pickers[pickerIdx];
//用選號器從集群當中隨機跳出一對來，待處理的<server,region>對
Pair<Pair<Integer, Integer>, Pair<Integer, Integer>> picks = p.pick(cluster);
int leftServer = picks.getFirst().getFirst();
int leftRegion = picks.getFirst().getSecond();
int rightServer = picks.getSecond().getFirst();
int rightRegion = picks.getSecond().getSecond();
cluster.moveOrSwapRegion(leftServer,
rightServer,
leftRegion,
rightRegion);
//移動或者交換完之后，看看新的開銷是否要繼續
newCost = computeCost(cluster, currentCost);
// Should this be kept? 挺好，保存新狀態
if (newCost < currentCost) {
currentCost = newCost;
} else {
// 操作不劃算，就回退
cluster.moveOrSwapRegion(leftServer,
rightServer,
rightRegion,
leftRegion);
}
if (initCost > currentCost) {
//找到了滿意的平衡方案
List<RegionPlan> plans = createRegionPlans(cluster);
return plans;
}
? 上面的被我清除了細枝末節之后的代碼主體，okay，上面邏輯過程如下：
1. 生成一個虛擬的集群cluster，方便計算計算當前狀態的開銷，其中clusterState是表的狀態，loads是整個集群的狀態。
// Keep track of servers to iterate through them. Cluster cluster = new Cluster(clusterState, loads, regionFinder); //計算出來當前的開銷 double currentCost = computeCost(cluster, Double.MAX_VALUE); double initCost = currentCost; double newCost = currentCost;?2. 然后循環computedMaxSteps次，隨機從選出一個picker來計算平衡方案
int pickerIdx = RANDOM.nextInt(pickers.length); RegionPicker p = pickers[pickerIdx]; //用選號器從集群當中隨機跳出一對來，待處理的<server,region>對 Pair<Pair<Integer, Integer>, Pair<Integer, Integer>> picks = p.pick(cluster);?
? picker是啥？這里面有三個，第一個是RandomRegionPicker是隨機挑選region，這里就不詳細介紹了，主要討論后面兩個；第二個LoadPicker是計算負載的，第三個主要是考慮本地性的。
給我感覺就很像ZF的搖號器一樣，用哪種算法還要搖個號
pickers = new RegionPicker[] {new RandomRegionPicker(),new LoadPicker(),localityPicker };?
? 下面我們先看localityPicker的pick方法，這個方法是隨機抽選出來一個server、region，找出region的其他本地機器，然后他們返回。
@OverridePair<Pair<Integer, Integer>, Pair<Integer, Integer>> pick(Cluster cluster) {if (this.masterServices == null) {return new Pair<Pair<Integer, Integer>, Pair<Integer, Integer>>(new Pair<Integer, Integer>(-1,-1),new Pair<Integer, Integer>(-1,-1));}// Pick a random region server 隨機選出一個server來int thisServer = pickRandomServer(cluster);// Pick a random region on this server 隨機選出regionint thisRegion = pickRandomRegion(cluster, thisServer, 0.0f);if (thisRegion == -1) {return new Pair<Pair<Integer, Integer>, Pair<Integer, Integer>>(new Pair<Integer, Integer>(-1,-1),new Pair<Integer, Integer>(-1,-1));}// Pick the server with the highest locality 找出本地性最高的目標serverint otherServer = pickHighestLocalityServer(cluster, thisServer, thisRegion);// pick an region on the other server to potentially swapint otherRegion = this.pickRandomRegion(cluster, otherServer, 0.5f);return new Pair<Pair<Integer, Integer>, Pair<Integer, Integer>>(new Pair<Integer, Integer>(thisServer,thisRegion),new Pair<Integer, Integer>(otherServer,otherRegion));}?
? okay，這個結束了，下面我們看看LoadPicker吧。
@OverridePair<Pair<Integer, Integer>, Pair<Integer, Integer>> pick(Cluster cluster) {cluster.sortServersByRegionCount();//先挑選出負載最高的serverint thisServer = pickMostLoadedServer(cluster, -1);//再選出除了負載最高的server之外負載最低的serverint otherServer = pickLeastLoadedServer(cluster, thisServer);Pair<Integer, Integer> regions = pickRandomRegions(cluster, thisServer, otherServer);return new Pair<Pair<Integer, Integer>, Pair<Integer, Integer>>(new Pair<Integer, Integer>(thisServer, regions.getFirst()),new Pair<Integer, Integer>(otherServer, regions.getSecond()));}?
這里的負載高和負載低是按照Server上面的region數來算的，而不是存儲文件啥的，選出負載最高和負載最低的時候，又隨機抽出region來返回了。
pick挑選的過程介紹完了，那么很明顯，計算才是重頭戲了，什么樣的region會導致計算出來的分數高低呢？
3. 重點在計算函數上?computeCost(cluster, Double.MAX_VALUE)?結果這個函數也超級簡單，哈哈
protected double computeCost(Cluster cluster, double previousCost) {double total = 0;for (CostFunction c:costFunctions) {if (c.getMultiplier() <= 0) {continue;}total += c.getMultiplier() * c.cost(cluster);if (total > previousCost) {return total;}}return total;}?
遍歷CostFunction，拿cost的加權平均和計算出來。
那costFunction里面都有啥呢？localityCost又出現了，看來本地性是一個很大的考慮的情況。
costFunctions = new CostFunction[]{new RegionCountSkewCostFunction(conf),new MoveCostFunction(conf),localityCost,new TableSkewCostFunction(conf),regionLoadFunctions[0],regionLoadFunctions[1],regionLoadFunctions[2],regionLoadFunctions[3], }; regionLoadFunctions = new CostFromRegionLoadFunction[] {
new ReadRequestCostFunction(conf),
new WriteRequestCostFunction(conf),
new MemstoreSizeCostFunction(conf),
new StoreFileCostFunction(conf)
};
?
可以看出來，里面真正看中硬盤內容大小的，只有一個StoreFileCostFunction，cost的計算方式有些區別，但都是一個0-1之間的數字，下面給出里面5個函數都用過的cost的函數。
//cost函數double max = ((count - 1) * mean) + (total - mean); for (double n : stats) {double diff = Math.abs(mean - n);totalCost += diff; }double scaled = scale(0, max, totalCost); return scaled;//scale函數 protected double scale(double min, double max, double value) {if (max == 0 || value == 0) {return 0;}return Math.max(0d, Math.min(1d, (value - min) / max)); }
?
經過分析吧，我覺得影響里面最后cost最大的是它的權重，下面給一下，這些function的默認權重。
RegionCountSkewCostFunction hbase.master.balancer.stochastic.regionCountCost ，默認值500MoveCostFunction hbase.master.balancer.stochastic.moveCost，默認值是100
localityCost hbase.master.balancer.stochastic.localityCost，默認值是25
TableSkewCostFunction hbase.master.balancer.stochastic.tableSkewCost，默認值是35
ReadRequestCostFunction hbase.master.balancer.stochastic.readRequestCost，默認值是5
WriteRequestCostFunction hbase.master.balancer.stochastic.writeRequestCost，默認值是5
MemstoreSizeCostFunction hbase.master.balancer.stochastic.memstoreSizeCost，默認值是5
StoreFileCostFunction hbase.master.balancer.stochastic.storefileSizeCost，默認值是5
Storefile的默認值是5，那么低。。。可以試著提高一下這個參數，使它在計算cost消耗的時候，產生更加正向的意義，效果不好說。
4. 根據虛擬的集群狀態生成RegionPlan，這里就不說了
List<RegionPlan> plans = createRegionPlans(cluster);?
源碼的分析完畢，要想減少存儲內容分布不均勻，可以試著考慮增加一個picker，這樣又不會缺少對其他條件的考慮，具體可以參考LoadPicker，復制它的實現再寫一個，在pickMostLoadedServer和pickLeastLoadedServer這兩個方法里面把考慮的條件改一下，以前的條件是Integer[] servers = cluster.serverIndicesSortedByRegionCount; 通過這個來查找一下負載最高和最低的server，那么現在我們要在Cluster里面增加一個Server ---> StoreFile大小的關系映射集合，但是這里面沒有,只有regionLoads，RegionLoad這個類有一個方法getStorefileSizeMB可以獲得StoreFile的大小，我們通過里面的region和server的映射regionIndexToServerIndex來最后計算出來這個映射關系即可，這個計算映射關系個過程放在Cluster的構造函數里面。
?
?
?
總結
以上是生活随笔為你收集整理的hbase源码系列（一）Balancer 负载均衡的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 关于招聘和面试
- 下一篇： 翟树卿：如何让数据挖掘助力精准化营销