本文主要是從HBase應用程序設計與開(kāi)發(fā)的角度，總結幾種常用的性能優(yōu)化方法。有關(guān)HBase系統配置級別的優(yōu)化，這里涉及的不多，這部分可以參考：淘寶Ken Wu同學(xué)的博客。
1. 表的設計
1.1 Pre-Creating Regions
默認情況下，在創(chuàng )建HBase表的時(shí)候會(huì )自動(dòng)創(chuàng )建一個(gè)region分區，當導入數據的時(shí)候，所有的HBase客戶(hù)端都向這一個(gè)region寫(xiě)數據，直到這個(gè)region足夠大了才進(jìn)行切分。一種可以加快批量寫(xiě)入速度的方法是通過(guò)預先創(chuàng )建一些空的regions，這樣當數據寫(xiě)入HBase時(shí)，會(huì )按照region分區情況，在集群內做數據的負載均衡。
有關(guān)預分區，詳情參見(jiàn)：Table Creation: Pre-Creating Regions，下面是一個(gè)例子：
public static boolean createTable(HBaseAdmin admin, HTableDescriptor table, byte[][] splits)throws IOException { try { admin.createTable(table, splits); return true; } catch (TableExistsException e) { logger.info("table " + table.getNameAsString() + " already exists"); // the table already exists... return false; }}public static byte[][] getHexSplits(String startKey, String endKey, int numRegions) { byte[][] splits = new byte[numRegions-1][]; BigInteger lowestKey = new BigInteger(startKey, 16); BigInteger highestKey = new BigInteger(endKey, 16); BigInteger range = highestKey.subtract(lowestKey); BigInteger regionIncrement = range.divide(BigInteger.valueOf(numRegions)); lowestKey = lowestKey.add(regionIncrement); for(int i=0; i < numRegions-1;i++) { BigInteger key = lowestKey.add(regionIncrement.multiply(BigInteger.valueOf(i))); byte[] b = String.format("%016x", key).getBytes(); splits[i] = b; } return splits;}1.2 Row Key
HBase中row key用來(lái)檢索表中的記錄，支持以下三種方式：
- 通過(guò)單個(gè)row key訪(fǎng)問(wèn)：即按照某個(gè)row key鍵值進(jìn)行g(shù)et操作；
- 通過(guò)row key的range進(jìn)行scan：即通過(guò)設置startRowKey和endRowKey，在這個(gè)范圍內進(jìn)行掃描；
- 全表掃描：即直接掃描整張表中所有行記錄。
在HBase中，row key可以是任意字符串，最大長(cháng)度64KB，實(shí)際應用中一般為10~100bytes，存為byte[]字節數組，一般設計成定長(cháng)的。
row key是按照字典序存儲，因此，設計row key時(shí)，要充分利用這個(gè)排序特點(diǎn)，將經(jīng)常一起讀取的數據存儲到一塊，將最近可能會(huì )被訪(fǎng)問(wèn)的數據放在一塊。
舉個(gè)例子：如果最近寫(xiě)入HBase表中的數據是最可能被訪(fǎng)問(wèn)的，可以考慮將時(shí)間戳作為row key的一部分，由于是字典序排序，所以可以使用Long.MAX_VALUE – timestamp作為row key，這樣能保證新寫(xiě)入的數據在讀取時(shí)可以被快速命中。
1.3 Column Family
不要在一張表里定義太多的column family。目前Hbase并不能很好的處理超過(guò)2~3個(gè)column family的表。因為某個(gè)column family在flush的時(shí)候，它鄰近的column family也會(huì )因關(guān)聯(lián)效應被觸發(fā)flush，最終導致系統產(chǎn)生更多的I/O。感興趣的同學(xué)可以對自己的HBase集群進(jìn)行實(shí)際測試，從得到的測試結果數據驗證一下。
1.4 In Memory
創(chuàng )建表的時(shí)候，可以通過(guò)HColumnDescriptor.setInMemory(true)將表放到RegionServer的緩存中，保證在讀取的時(shí)候被cache命中。
1.5 Max Version
創(chuàng )建表的時(shí)候，可以通過(guò)HColumnDescriptor.setMaxVersions(int maxVersions)設置表中數據的最大版本，如果只需要保存最新版本的數據，那么可以設置setMaxVersions(1)。
1.6 Time To Live
創(chuàng )建表的時(shí)候，可以通過(guò)HColumnDescriptor.setTimeToLive(int timeToLive)設置表中數據的存儲生命期，過(guò)期數據將自動(dòng)被刪除，例如如果只需要存儲最近兩天的數據，那么可以設置setTimeToLive(2 * 24 * 60 * 60)。
1.7 Compact & Split
在HBase中，數據在更新時(shí)首先寫(xiě)入WAL 日志(HLog)和內存(MemStore)中，MemStore中的數據是排序的，當MemStore累計到一定閾值時(shí)，就會(huì )創(chuàng )建一個(gè)新的MemStore，并且將老的MemStore添加到flush隊列，由單獨的線(xiàn)程flush到磁盤(pán)上，成為一個(gè)StoreFile。于此同時(shí)， 系統會(huì )在zookeeper中記錄一個(gè)redo point，表示這個(gè)時(shí)刻之前的變更已經(jīng)持久化了(minor compact)。
StoreFile是只讀的，一旦創(chuàng )建后就不可以再修改。因此Hbase的更新其實(shí)是不斷追加的操作。當一個(gè)Store中的StoreFile達到一定的閾值后，就會(huì )進(jìn)行一次合并(major compact)，將對同一個(gè)key的修改合并到一起，形成一個(gè)大的StoreFile，當StoreFile的大小達到一定閾值后，又會(huì )對 StoreFile進(jìn)行分割(split)，等分為兩個(gè)StoreFile。
由于對表的更新是不斷追加的，處理讀請求時(shí)，需要訪(fǎng)問(wèn)Store中全部的StoreFile和MemStore，將它們按照row key進(jìn)行合并，由于StoreFile和MemStore都是經(jīng)過(guò)排序的，并且StoreFile帶有內存中索引，通常合并過(guò)程還是比較快的。
實(shí)際應用中，可以考慮必要時(shí)手動(dòng)進(jìn)行major compact，將同一個(gè)row key的修改進(jìn)行合并形成一個(gè)大的StoreFile。同時(shí)，可以將StoreFile設置大些，減少split的發(fā)生。
2. 寫(xiě)表操作
2.1 多HTable并發(fā)寫(xiě)
創(chuàng )建多個(gè)HTable客戶(hù)端用于寫(xiě)操作，提高寫(xiě)數據的吞吐量，一個(gè)例子：
static final Configuration conf = HBaseConfiguration.create();static final String table_log_name = “user_log”;wTableLog = new HTable[tableN];for (int i = 0; i < tableN; i++) { wTableLog[i] = new HTable(conf, table_log_name); wTableLog[i].setWriteBufferSize(5 * 1024 * 1024); //5MB wTableLog[i].setAutoFlush(false);}2.2 HTable參數設置
2.2.1 Auto Flush
通過(guò)調用HTable.setAutoFlush(false)方法可以將HTable寫(xiě)客戶(hù)端的自動(dòng)flush關(guān)閉，這樣可以批量寫(xiě)入數據到HBase，而不是有一條put就執行一次更新，只有當put填滿(mǎn)客戶(hù)端寫(xiě)緩存時(shí)，才實(shí)際向HBase服務(wù)端發(fā)起寫(xiě)請求。默認情況下auto flush是開(kāi)啟的。
2.2.2 Write Buffer
通過(guò)調用HTable.setWriteBufferSize(writeBufferSize)方法可以設置HTable客戶(hù)端的寫(xiě)buffer大小，如果新設置的buffer小于當前寫(xiě)buffer中的數據時(shí)，buffer將會(huì )被flush到服務(wù)端。其中，writeBufferSize的單位是byte字節數，可以根據實(shí)際寫(xiě)入數據量的多少來(lái)設置該值。
2.2.3 WAL Flag
在HBae中，客戶(hù)端向集群中的RegionServer提交數據時(shí)（Put/Delete操作），首先會(huì )先寫(xiě)WAL（Write Ahead Log）日志（即HLog，一個(gè)RegionServer上的所有Region共享一個(gè)HLog），只有當WAL日志寫(xiě)成功后，再接著(zhù)寫(xiě)MemStore，然后客戶(hù)端被通知提交數據成功；如果寫(xiě)WAL日志失敗，客戶(hù)端則被通知提交失敗。這樣做的好處是可以做到RegionServer宕機后的數據恢復。
因此，對于相對不太重要的數據，可以在Put/Delete操作時(shí)，通過(guò)調用Put.setWriteToWAL(false)或Delete.setWriteToWAL(false)函數，放棄寫(xiě)WAL日志，從而提高數據寫(xiě)入的性能。
值得注意的是：謹慎選擇關(guān)閉WAL日志，因為這樣的話(huà)，一旦RegionServer宕機，Put/Delete的數據將會(huì )無(wú)法根據WAL日志進(jìn)行恢復。
2.3 批量寫(xiě)
通過(guò)調用HTable.put(Put)方法可以將一個(gè)指定的row key記錄寫(xiě)入HBase，同樣HBase提供了另一個(gè)方法：通過(guò)調用HTable.put(List<Put>)方法可以將指定的row key列表，批量寫(xiě)入多行記錄，這樣做的好處是批量執行，只需要一次網(wǎng)絡(luò )I/O開(kāi)銷(xiāo)，這對于對數據實(shí)時(shí)性要求高，網(wǎng)絡(luò )傳輸RTT高的情景下可能帶來(lái)明顯的性能提升。
2.4 多線(xiàn)程并發(fā)寫(xiě)
在客戶(hù)端開(kāi)啟多個(gè)HTable寫(xiě)線(xiàn)程，每個(gè)寫(xiě)線(xiàn)程負責一個(gè)HTable對象的flush操作，這樣結合定時(shí)flush和寫(xiě)buffer（writeBufferSize），可以既保證在數據量小的時(shí)候，數據可以在較短時(shí)間內被flush（如1秒內），同時(shí)又保證在數據量大的時(shí)候，寫(xiě)buffer一滿(mǎn)就及時(shí)進(jìn)行flush。下面給個(gè)具體的例子：
for (int i = 0; i < threadN; i++) { Thread th = new Thread() { public void run() { while (true) { try { sleep(1000); //1 second } catch (InterruptedException e) { e.printStackTrace(); } synchronized (wTableLog[i]) { try { wTableLog[i].flushCommits(); } catch (IOException e) { e.printStackTrace(); } } } } }; th.setDaemon(true); th.start();}3. 讀表操作
3.1 多HTable并發(fā)讀
創(chuàng )建多個(gè)HTable客戶(hù)端用于讀操作，提高讀數據的吞吐量，一個(gè)例子：
static final Configuration conf = HBaseConfiguration.create();static final String table_log_name = “user_log”;rTableLog = new HTable[tableN];for (int i = 0; i < tableN; i++) { rTableLog[i] = new HTable(conf, table_log_name); rTableLog[i].setScannerCaching(50);}<!--[if gte mso 9]> Normal 0 7.8 磅 0 2 false false false EN-US ZH-CN X-NONE <![endif]--><!--[if gte mso 9]> <![endif]-->
3.2 HTable參數設置
3.2.1 Scanner Caching
通過(guò)調用HTable.setScannerCaching(int scannerCaching)可以設置HBase scanner一次從服務(wù)端抓取的數據條數，默認情況下一次一條。通過(guò)將此值設置成一個(gè)合理的值，可以減少scan過(guò)程中next()的時(shí)間開(kāi)銷(xiāo)，代價(jià)是scanner需要通過(guò)客戶(hù)端的內存來(lái)維持這些被cache的行記錄。
3.2.2 Scan Attribute Selection
scan時(shí)指定需要的Column Family，可以減少網(wǎng)絡(luò )傳輸數據量，否則默認scan操作會(huì )返回整行所有Column Family的數據。
3.2.3 Close ResultScanner
通過(guò)scan取完數據后，記得要關(guān)閉ResultScanner，否則RegionServer可能會(huì )出現問(wèn)題（對應的Server資源無(wú)法釋放）。
3.3 批量讀
通過(guò)調用HTable.get(Get)方法可以根據一個(gè)指定的row key獲取一行記錄，同樣HBase提供了另一個(gè)方法：通過(guò)調用HTable.get(List<Get>)方法可以根據一個(gè)指定的row key列表，批量獲取多行記錄，這樣做的好處是批量執行，只需要一次網(wǎng)絡(luò )I/O開(kāi)銷(xiāo)，這對于對數據實(shí)時(shí)性要求高而且網(wǎng)絡(luò )傳輸RTT高的情景下可能帶來(lái)明顯的性能提升。
3.4 多線(xiàn)程并發(fā)讀
在客戶(hù)端開(kāi)啟多個(gè)HTable讀線(xiàn)程，每個(gè)讀線(xiàn)程負責通過(guò)HTable對象進(jìn)行get操作。下面是一個(gè)多線(xiàn)程并發(fā)讀取HBase，獲取店鋪一天內各分鐘PV值的例子：
public class DataReaderServer { //獲取店鋪一天內各分鐘PV值的入口函數 public static ConcurrentHashMap getUnitMinutePV(long uid, long startStamp, long endStamp){ long min = startStamp; int count = (int)((endStamp - startStamp) / (60*1000)); List lst = new ArrayList(); for (int i = 0; i <= count; i++) { min = startStamp + i * 60 * 1000; lst.add(uid + "_" + min); } return parallelBatchMinutePV(lst); } //多線(xiàn)程并發(fā)查詢(xún)，獲取分鐘PV值private static ConcurrentHashMap parallelBatchMinutePV(List lstKeys){ ConcurrentHashMap hashRet = new ConcurrentHashMap(); int parallel = 3; List<List<String>> lstBatchKeys = null; if (lstKeys.size() < parallel ){ lstBatchKeys = new ArrayList<List<String>>(1); lstBatchKeys.add(lstKeys); } else{ lstBatchKeys = new ArrayList<List<String>>(parallel); for(int i = 0; i < parallel; i++ ){ List lst = new ArrayList(); lstBatchKeys.add(lst); } for(int i = 0 ; i < lstKeys.size() ; i ++ ){ lstBatchKeys.get(i%parallel).add(lstKeys.get(i)); } } List >> futures = new ArrayList >>(5); ThreadFactoryBuilder builder = new ThreadFactoryBuilder(); builder.setNameFormat("ParallelBatchQuery"); ThreadFactory factory = builder.build(); ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(lstBatchKeys.size(), factory); for(List keys : lstBatchKeys){ Callable< ConcurrentHashMap > callable = new BatchMinutePVCallable(keys); FutureTask< ConcurrentHashMap > future = (FutureTask< ConcurrentHashMap >) executor.submit(callable); futures.add(future); } executor.shutdown(); // Wait for all the tasks to finish try { boolean stillRunning = !executor.awaitTermination( 5000000, TimeUnit.MILLISECONDS); if (stillRunning) { try { executor.shutdownNow(); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } } } catch (InterruptedException e) { try { Thread.currentThread().interrupt(); } catch (Exception e1) { // TODO Auto-generated catch block e1.printStackTrace(); } } // Look for any exception for (Future f : futures) { try { if(f.get() != null) { hashRet.putAll((ConcurrentHashMap)f.get()); } } catch (InterruptedException e) { try { Thread.currentThread().interrupt(); } catch (Exception e1) { // TODO Auto-generated catch block e1.printStackTrace(); } } catch (ExecutionException e) { e.printStackTrace(); } } return hashRet; } //一個(gè)線(xiàn)程批量查詢(xún)，獲取分鐘PV值 protected static ConcurrentHashMap getBatchMinutePV(List lstKeys){ ConcurrentHashMap hashRet = null; List lstGet = new ArrayList(); String[] splitValue = null; for (String s : lstKeys) { splitValue = s.split("_"); long uid = Long.parseLong(splitValue[0]); long min = Long.parseLong(splitValue[1]); byte[] key = new byte[16]; Bytes.putLong(key, 0, uid); Bytes.putLong(key, 8, min); Get g = new Get(key); g.addFamily(fp); lstGet.add(g); } Result[] res = null; try { res = tableMinutePV[rand.nextInt(tableN)].get(lstGet); } catch (IOException e1) { logger.error("tableMinutePV exception, e=" + e1.getStackTrace()); } if (res != null && res.length > 0) { hashRet = new ConcurrentHashMap(res.length); for (Result re : res) { if (re != null && !re.isEmpty()) { try { byte[] key = re.getRow(); byte[] value = re.getValue(fp, cp); if (key != null && value != null) { hashRet.put(String.valueOf(Bytes.toLong(key, Bytes.SIZEOF_LONG)), String.valueOf(Bytes .toLong(value))); } } catch (Exception e2) { logger.error(e2.getStackTrace()); } } } } return hashRet; }}//調用接口類(lèi)，實(shí)現Callable接口class BatchMinutePVCallable implements Callable>{ private List keys; public BatchMinutePVCallable(List lstKeys ) { this.keys = lstKeys; } public ConcurrentHashMap call() throws Exception { return DataReadServer.getBatchMinutePV(keys); }}<!--[if gte mso 9]> Normal 0 7.8 磅 0 2 false false false EN-US ZH-CN X-NONE <![endif]--><!--[if gte mso 9]> <![endif]-->
3.5 緩存查詢(xún)結果
對于頻繁查詢(xún)HBase的應用場(chǎng)景，可以考慮在應用程序中做緩存，當有新的查詢(xún)請求時(shí)，首先在緩存中查找，如果存在則直接返回，不再查詢(xún)HBase；否則對HBase發(fā)起讀請求查詢(xún)，然后在應用程序中將查詢(xún)結果緩存起來(lái)。至于緩存的替換策略，可以考慮LRU等常用的策略。
3.6 Blockcache
HBase上Regionserver的內存分為兩個(gè)部分，一部分作為Memstore，主要用來(lái)寫(xiě)；另外一部分作為BlockCache，主要用于讀。
寫(xiě)請求會(huì )先寫(xiě)入Memstore，Regionserver會(huì )給每個(gè)region提供一個(gè)Memstore，當Memstore滿(mǎn)64MB以后，會(huì )啟動(dòng) flush刷新到磁盤(pán)。當Memstore的總大小超過(guò)限制時(shí)（heapsize * hbase.regionserver.global.memstore.upperLimit * 0.9），會(huì )強行啟動(dòng)flush進(jìn)程，從最大的Memstore開(kāi)始flush直到低于限制。
讀請求先到Memstore中查數據，查不到就到BlockCache中查，再查不到就會(huì )到磁盤(pán)上讀，并把讀的結果放入BlockCache。由于BlockCache采用的是LRU策略，因此BlockCache達到上限(heapsize * hfile.block.cache.size * 0.85)后，會(huì )啟動(dòng)淘汰機制，淘汰掉最老的一批數據。
一個(gè)Regionserver上有一個(gè)BlockCache和N個(gè)Memstore，它們的大小之和不能大于等于heapsize * 0.8，否則HBase不能啟動(dòng)。默認BlockCache為0.2，而Memstore為0.4。對于注重讀響應時(shí)間的系統，可以將 BlockCache設大些，比如設置BlockCache=0.4，Memstore=0.39，以加大緩存的命中率。
有關(guān)BlockCache機制，請參考這里：HBase的Block cache，HBase的blockcache機制，hbase中的緩存的計算與使用。
4. 數據計算
4.1 服務(wù)端計算
Coprocessor運行于HBase RegionServer服務(wù)端，各個(gè)Regions保持對與其相關(guān)的coprocessor實(shí)現類(lèi)的引用，coprocessor類(lèi)可以通過(guò)RegionServer上classpath中的本地jar或HDFS的classloader進(jìn)行加載。
目前，已提供有幾種coprocessor：
- Coprocessor：提供對于region管理的鉤子，例如region的open/close/split/flush/compact等；
- RegionObserver：提供用于從客戶(hù)端監控表相關(guān)操作的鉤子，例如表的get/put/scan/delete等；
- Endpoint：提供可以在region上執行任意函數的命令觸發(fā)器。一個(gè)使用例子是RegionServer端的列聚合，這里有代碼示例。
以上只是有關(guān)coprocessor的一些基本介紹，本人沒(méi)有對其實(shí)際使用的經(jīng)驗，對它的可用性和性能數據不得而知。感興趣的同學(xué)可以嘗試一下，歡迎討論。
4.2 寫(xiě)端計算
4.2.1 計數
HBase本身可以看作是一個(gè)可以水平擴展的Key-Value存儲系統，但是其本身的計算能力有限（Coprocessor可以提供一定的服務(wù)端計算），因此，使用HBase時(shí)，往往需要從寫(xiě)端或者讀端進(jìn)行計算，然后將最終的計算結果返回給調用者。舉兩個(gè)簡(jiǎn)單的例子：
- PV計算：通過(guò)在HBase寫(xiě)端內存中，累加計數，維護PV值的更新，同時(shí)為了做到持久化，定期（如1秒）將PV計算結果同步到HBase中，這樣查詢(xún)端最多會(huì )有1秒鐘的延遲，能看到秒級延遲的PV結果。
- 分鐘PV計算：與上面提到的PV計算方法相結合，每分鐘將當前的累計PV值，按照rowkey + minute作為新的rowkey寫(xiě)入HBase中，然后在查詢(xún)端通過(guò)scan得到當天各個(gè)分鐘以前的累計PV值，然后順次將前后兩分鐘的累計PV值相減，就得到了當前一分鐘內的PV值，從而最終也就得到當天各個(gè)分鐘內的PV值。
4.2.2 去重
對于UV的計算，就是個(gè)去重計算的例子。分兩種情況：
- 如果內存可以容納，那么可以在Hash表中維護所有已經(jīng)存在的UV標識，每當新來(lái)一個(gè)標識時(shí)，通過(guò)快速查找Hash確定是否是一個(gè)新的UV，若是則UV值加1，否則UV值不變。另外，為了做到持久化或提供給查詢(xún)接口使用，可以定期（如1秒）將UV計算結果同步到HBase中。
- 如果內存不能容納，可以考慮采用Bloom Filter來(lái)實(shí)現，從而盡可能的減少內存的占用情況。除了UV的計算外，判斷URL是否存在也是個(gè)典型的應用場(chǎng)景。
4.3 讀端計算
如果對于響應時(shí)間要求比較苛刻的情況（如單次http請求要在毫秒級時(shí)間內返回），個(gè)人覺(jué)得讀端不宜做過(guò)多復雜的計算邏輯，盡量做到讀端功能單一化：即從HBase RegionServer讀到數據（scan或get方式）后，按照數據格式進(jìn)行簡(jiǎn)單的拼接，直接返回給前端使用。當然，如果對于響應時(shí)間要求一般，或者業(yè)務(wù)特點(diǎn)需要，也可以在讀端進(jìn)行一些計算邏輯。
5. 總結
作為一個(gè)Key-Value存儲系統，HBase并不是萬(wàn)能的，它有自己獨特的地方。因此，基于它來(lái)做應用時(shí)，我們往往需要從多方面進(jìn)行優(yōu)化改進(jìn)（表設計、讀表操作、寫(xiě)表操作、數據計算等），有時(shí)甚至還需要從系統級對HBase進(jìn)行配置調優(yōu)，更甚至可以對HBase本身進(jìn)行優(yōu)化。這屬于不同的層次范疇。
總之，概括來(lái)講，對系統進(jìn)行優(yōu)化時(shí)，首先定位到影響你的程序運行性能的瓶頸之處，然后有的放矢進(jìn)行針對行的優(yōu)化。如果優(yōu)化后滿(mǎn)足你的期望，那么就可以停止優(yōu)化；否則繼續尋找新的瓶頸之處，開(kāi)始新的優(yōu)化，直到滿(mǎn)足性能要求。
以上就是從項目開(kāi)發(fā)中總結的一點(diǎn)經(jīng)驗，如有不對之處，歡迎大家不吝賜教。
轉載文章請注明，轉載自：量子數科院[http://www.linezing.com/blog]。文章均為原創(chuàng )，版權歸量子統計所有
