hbase熱點問題(數據傾斜)解決方案---rowkey散列和預分區設計

 Hbase的表會被劃分爲1....n個Region,被託管在RegionServer中。Region二個重要的屬性：Startkey與EndKey表示這個Region維護的rowkey的範圍，當咱們要讀寫數據時，若是rowkey落在某個start-end key範圍內，那麼就會定位到目標region而且讀寫到相關的數據。

    默認狀況下，當咱們經過hbaseAdmin指定TableDescriptor來建立一張表時，只有一個region正處於混沌時期，start-end key無邊界，可謂海納百川。全部的rowkey都寫入到這個region裏，而後數據愈來愈多，region的size愈來愈大時，大到必定的閥值，hbase就會將region一分爲二，成爲2個region，這個過程稱爲分裂（region-split）。

    若是咱們就這樣默認建表，表裏不斷的put數據，更嚴重的是咱們的rowkey仍是順序增大的，是比較可怕的。存在的缺點比較明顯：首先是熱點寫，咱們老是向最大的start key所在的region寫數據，由於咱們的rowkey老是會比以前的大，而且hbase的是按升序方式排序的。因此寫操做老是被定位到無上界的那個region中；其次，因爲熱點，咱們老是往最大的start key的region寫記錄，以前分裂出來的region不會被寫數據，有點打入冷宮的感受，他們都處於半滿狀態，這樣的分佈也是不利的。

    若是在寫比較頻繁的場景下，數據增加太快，split的次數也會增多，因爲split是比較耗費資源的，因此咱們並不但願這種事情常常發生。

    在集羣中爲了獲得更好的並行性，咱們但願有好的load blance，讓每一個節點提供的請求都是均衡的，咱們也不但願，region不要常常split，由於split會使server有一段時間的停頓，如何能作到呢？

    隨機散列與預分區兩者結合起來，是比較完美的。預分區一開始就預建好了一部分region，這些region都維護着本身的start-end keys，在配合上隨機散列，寫數據能均衡的命中這些預建的region，就能解決上面的那些缺點，大大提供性能。

1、解決思路

    提供兩種思路：hash與partition。

一、hash方案

    hash就是rowkey前面由一串隨機字符串組成，隨機字符串生成方式能夠由SHA或者MD5方式生成，只要region所管理的start-end keys範圍比較隨機，那麼就能夠解決寫熱點問題。例如：

Java代碼

 

1. long currentId = 1L;  
2. byte [] rowkey = Bytes.add(MD5Hash.getMD5AsHex(Bytes.toBytes(currentId))  
3.                     .substring(0, 8).getBytes(),Bytes.toBytes(currentId));  

     假如rowkey本來是自增加的long型，能夠將rowkey轉爲hash再轉爲bytes，加上自己id轉爲bytes，這樣就生成隨便的rowkey。那麼對於這種方式的rowkey設計，如何去進行預分區呢？

1. 取樣，先隨機生成必定數量的rowkey，將取樣數據按升序排序放到一個集合裏。
2. 根據預分區的region個數，對整個集合平均分割，便是相關的splitkeys。
3. HBaseAdmin.createTable(HTableDescriptor tableDescriptor,byte[][] splitkeys)能夠指定預分區的splitkey，即指定region間的rowkey臨界值。

    建立split計算器，用於從抽樣數據生成一個比較合適的splitkeys

Java代碼

 

1. public class HashChoreWoker implements SplitKeysCalculator{  
2.     //隨機取機數目  
3.     private int baseRecord;  
4.     //rowkey生成器  
5.     private RowKeyGenerator rkGen;  
6.     //取樣時，由取樣數目及region數相除所得的數量.  
7.     private int splitKeysBase;  
8.     //splitkeys個數  
9.     private int splitKeysNumber;  
10.     //由抽樣計算出來的splitkeys結果  
11.     private byte[][] splitKeys;  
12.   
13.     public HashChoreWoker(int baseRecord, int prepareRegions) {  
14.         this.baseRecord = baseRecord;  
15.         //實例化rowkey生成器  
16.         rkGen = new HashRowKeyGenerator();  
17.         splitKeysNumber = prepareRegions - 1;  
18.         splitKeysBase = baseRecord / prepareRegions;  
19.     }  
20.   
21.     public byte[][] calcSplitKeys() {  
22.         splitKeys = new byte[splitKeysNumber][];  
23.         //使用treeset保存抽樣數據，已排序過  
24.         TreeSet<byte[]> rows = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);  
25.         for (int i = 0; i < baseRecord; i++) {  
26.             rows.add(rkGen.nextId());  
27.         }  
28.         int pointer = 0;  
29.         Iterator<byte[]> rowKeyIter = rows.iterator();  
30.         int index = 0;  
31.         while (rowKeyIter.hasNext()) {  
32.             byte[] tempRow = rowKeyIter.next();  
33.             rowKeyIter.remove();  
34.             if ((pointer != 0) && (pointer % splitKeysBase == 0)) {  
35.                 if (index < splitKeysNumber) {  
36.                     splitKeys[index] = tempRow;  
37.                     index ++;  
38.                 }  
39.             }  
40.             pointer ++;  
41.         }  
42.         rows.clear();  
43.         rows = null;  
44.         return splitKeys;  
45.     }  
46. }  

     KeyGenerator及實現

Java代碼

 

1. //interface  
2. public interface RowKeyGenerator {  
3.     byte [] nextId();  
4. }  
5. //implements  
6. public class HashRowKeyGenerator implements RowKeyGenerator {  
7.     private long currentId = 1;  
8.     private long currentTime = System.currentTimeMillis();  
9.     private Random random = new Random();  
10.     public byte[] nextId() {  
11.         try {  
12.             currentTime += random.nextInt(1000);  
13.             byte[] lowT = Bytes.copy(Bytes.toBytes(currentTime), 4, 4);  
14.             byte[] lowU = Bytes.copy(Bytes.toBytes(currentId), 4, 4);  
15.             return Bytes.add(MD5Hash.getMD5AsHex(Bytes.add(lowU, lowT)).substring(0, 8).getBytes(),  
16.                     Bytes.toBytes(currentId));  
17.         } finally {  
18.             currentId++;  
19.         }  
20.     }  
21. }  

     unit test case測試

Java代碼

 

1. @Test  
2. public void testHashAndCreateTable() throws Exception{  
3.         HashChoreWoker worker = new HashChoreWoker(1000000,10);  
4.         byte [][] splitKeys = worker.calcSplitKeys();  
5.           
6.         HBaseAdmin admin = new HBaseAdmin(HBaseConfiguration.create());  
7.         TableName tableName = TableName.valueOf("hash_split_table");  
8.           
9.         if (admin.tableExists(tableName)) {  
10.             try {  
11.                 admin.disableTable(tableName);  
12.             } catch (Exception e) {  
13.             }  
14.             admin.deleteTable(tableName);  
15.         }  
16.   
17.         HTableDescriptor tableDesc = new HTableDescriptor(tableName);  
18.         HColumnDescriptor columnDesc = new HColumnDescriptor(Bytes.toBytes("info"));  
19.         columnDesc.setMaxVersions(1);  
20.         tableDesc.addFamily(columnDesc);  
21.   
22.         admin.createTable(tableDesc ,splitKeys);  
23.   
24.         admin.close();  
25.     }  

     查看建表結果，執行：scan 'hbase:meta'

    以上咱們只是顯示了部分region的信息，能夠看到region的start-end key仍是比較隨機散列的。一樣能夠查看hdfs的目錄結構，的確和預期的38個預分區一致： 

    以上就是按照hash方式，預建好分區，之後再插入數據的時候，也是按照此rowkeyGenerator的方式生成rowkey。

二、partition的方式

    partition顧名思義就是分區式，這種分區有點相似於mapreduce中的partitioner，將區域用長整數做爲分區號，每一個region管理着相應的區域數據，在rowkey生成時，將ID取模後，而後拼上ID總體做爲rowkey，這個比較簡單，不須要取樣，splitkeys也很是簡單，直接是分區號便可。直接上代碼：

Java代碼

 

1. public class PartitionRowKeyManager implements RowKeyGenerator,  
2.         SplitKeysCalculator {  
3.   
4.     public static final int DEFAULT_PARTITION_AMOUNT = 20;  
5.     private long currentId = 1;  
6.     private int partition = DEFAULT_PARTITION_AMOUNT;  
7.     public void setPartition(int partition) {  
8.         this.partition = partition;  
9.     }  
10.   
11.     public byte[] nextId() {  
12.         try {  
13.             long partitionId = currentId % partition;  
14.             return Bytes.add(Bytes.toBytes(partitionId),  
15.                     Bytes.toBytes(currentId));  
16.         } finally {  
17.             currentId++;  
18.         }  
19.     }  
20.   
21.     public byte[][] calcSplitKeys() {  
22.         byte[][] splitKeys = new byte[partition - 1][];  
23.         for(int i = 1; i < partition ; i ++) {  
24.             splitKeys[i-1] = Bytes.toBytes((long)i);  
25.         }  
26.         return splitKeys;  
27.     }  
28. }  

    calcSplitKeys方法比較單純，splitkey就是partition的編號，測試類以下：

Java代碼

 

1. @Test  
2.     public void testPartitionAndCreateTable() throws Exception{  
3.           
4.         PartitionRowKeyManager rkManager = new PartitionRowKeyManager();  
5.         //只預建10個分區  
6.         rkManager.setPartition(10);  
7.           
8.         byte [][] splitKeys = rkManager.calcSplitKeys();  
9.           
10.         HBaseAdmin admin = new HBaseAdmin(HBaseConfiguration.create());  
11.         TableName tableName = TableName.valueOf("partition_split_table");  
12.           
13.         if (admin.tableExists(tableName)) {  
14.             try {  
15.                 admin.disableTable(tableName);  
16.   
17.             } catch (Exception e) {  
18.             }  
19.             admin.deleteTable(tableName);  
20.         }  
21.   
22.         HTableDescriptor tableDesc = new HTableDescriptor(tableName);  
23.         HColumnDescriptor columnDesc = new HColumnDescriptor(Bytes.toBytes("info"));  
24.         columnDesc.setMaxVersions(1);  
25.         tableDesc.addFamily(columnDesc);  
26.   
27.         admin.createTable(tableDesc ,splitKeys);  
28.   
29.         admin.close();  
30.     }  

     一樣咱們能夠看看meta表和hdfs的目錄結果，其實和hash相似，region都會分好區。

     經過partition實現的loadblance寫的話，固然生成rowkey方式也要結合當前的region數目取模而求得，你們一樣也能夠作些實驗，看看數據插入後的分佈。

     在這裏也順提一下，若是是順序的增加型原id,能夠將id保存到一個數據庫，傳統的也好,redis的也好，每次取的時候，將數值設大1000左右，之後id能夠在內存內增加，當內存數量已經超過1000的話，再去load下一個，有點相似於oracle中的sqeuence.

     隨機分佈加預分區也不是一勞永逸的。由於數據是不斷地增加的，隨着時間不斷地推移，已經分好的區域，或許已經裝不住更多的數據，固然就要進一步進行split了，一樣也會出現性能損耗問題，因此咱們仍是要規劃好數據增加速率，觀察好數據按期維護，按需分析是否要進一步分行手工將分區再分好，也或者是更嚴重的是新建表，作好更大的預分區而後進行數據遷移。若是數據裝不住了，對於partition方式預分區的話，若是讓它天然分裂的話，狀況分嚴重一點。由於分裂出來的分區號會是同樣的，因此計算到partitionId的話，其實仍是回到了順序寫年代，會有部分熱點寫問題出現，若是使用partition方式生成主鍵的話，數據增加後就要不斷地調整分區了，好比增多預分區，或者加入子分區號的處理.(咱們的分區號爲long型，能夠將它做爲多級partition)

    以上基本已經講完了防止熱點寫使用的方法和防止頻繁split而採起的預分區。但rowkey設計，遠遠也不止這些，好比rowkey長度，而後它的長度最大能夠爲char的MAXVALUE,可是看過以前我寫KeyValue的分析知道，咱們的數據都是以KeyValue方式存儲在MemStore或者HFile中的，每一個KeyValue都會存儲rowKey的信息，若是rowkey太大的話，好比是128個字節，一行10個字段的表，100萬行記錄，光rowkey就佔了1.2G+因此長度仍是不要過長，另外設計，仍是按需求來吧。