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+因此长度仍是不要过长，另外设计，仍是按需求来吧。