paulwong

          My Links

          Blog Stats

          常用鏈接

          留言簿(67)

          隨筆分類(1393)

          隨筆檔案(1151)

          文章分類(7)

          文章檔案(10)

          相冊

          收藏夾(2)

          AI

          Develop

          E-BOOK

          Other

          養生

          微服務

          搜索

          最新評論

          閱讀排行榜

          評論排行榜

          60天內閱讀排行

          HBase性能優化方法總結


          本文主要是從HBase應用程序設計與開發的角度,總結幾種常用的性能優化方法。有關HBase系統配置級別的優化,這里涉及的不多,這部分可以參考:淘寶Ken Wu同學的博客。

          1. 表的設計
          1.1 Pre-Creating Regions
          默認情況下,在創建HBase表的時候會自動創建一個region分區,當導入數據的時候,所有的HBase客戶端都向這一個region寫數據,直到這個region足夠大了才進行切分。一種可以加快批量寫入速度的方法是通過預先創建一些空的regions,這樣當數據寫入HBase時,會按照region分區情況,在集群內做數據的負載均衡。

          有關預分區,詳情參見:Table Creation: Pre-Creating Regions,下面是一個例子:
          publicstaticbooleancreateTable(HBaseAdmin admin, HTableDescriptor table,byte[][] splits)
          throwsIOException {
            try{
              admin.createTable(table, splits);
              returntrue;
            }catch(TableExistsException e) {
              logger.info("table "+ table.getNameAsString() +" already exists");
              // the table already exists
              returnfalse;
            }
          }
           
          publicstaticbyte[][] getHexSplits(String startKey, String endKey,intnumRegions) {
            byte[][] splits =newbyte[numRegions-1][];
            BigInteger lowestKey =newBigInteger(startKey,16);
            BigInteger highestKey =newBigInteger(endKey,16);
            BigInteger range = highestKey.subtract(lowestKey);
            BigInteger regionIncrement = range.divide(BigInteger.valueOf(numRegions));
            lowestKey = lowestKey.add(regionIncrement);
            for(inti=0; i < numRegions-1;i++) {
              BigInteger key = lowestKey.add(regionIncrement.multiply(BigInteger.valueOf(i)));
              byte[] b = String.format("%016x", key).getBytes();
              splits[i] = b;
            }
            returnsplits;
          }

          1.2 Row Key
          HBase中row key用來檢索表中的記錄,支持以下三種方式:

          通過單個row key訪問:即按照某個row key鍵值進行get操作;
          通過row key的range進行scan:即通過設置startRowKey和endRowKey,在這個范圍內進行掃描;
          全表掃描:即直接掃描整張表中所有行記錄。
          在HBase中,row key可以是任意字符串,最大長度64KB,實際應用中一般為10~100bytes,存為byte[]字節數組,一般設計成定長的。

          row key是按照字典序存儲,因此,設計row key時,要充分利用這個排序特點,將經常一起讀取的數據存儲到一塊,將最近可能會被訪問的數據放在一塊。

          舉個例子:如果最近寫入HBase表中的數據是最可能被訪問的,可以考慮將時間戳作為row key的一部分,由于是字典序排序,所以可以使用Long.MAX_VALUE – timestamp作為row key,這樣能保證新寫入的數據在讀取時可以被快速命中。

          1.3 Column Family
          不要在一張表里定義太多的column family。目前Hbase并不能很好的處理超過2~3個column family的表。因為某個column family在flush的時候,它鄰近的column family也會因關聯效應被觸發flush,最終導致系統產生更多的I/O。感興趣的同學可以對自己的HBase集群進行實際測試,從得到的測試結果數據驗證一下。

          1.4 In Memory
          創建表的時候,可以通過HColumnDescriptor.setInMemory(true)將表放到RegionServer的緩存中,保證在讀取的時候被cache命中。

          1.5 Max Version
          創建表的時候,可以通過HColumnDescriptor.setMaxVersions(int maxVersions)設置表中數據的最大版本,如果只需要保存最新版本的數據,那么可以設置setMaxVersions(1)。

          1.6 Time To Live
          創建表的時候,可以通過HColumnDescriptor.setTimeToLive(int timeToLive)設置表中數據的存儲生命期,過期數據將自動被刪除,例如如果只需要存儲最近兩天的數據,那么可以設置setTimeToLive(2 * 24 * 60 * 60)。

          1.7 Compact & Split
          在HBase中,數據在更新時首先寫入WAL 日志(HLog)和內存(MemStore)中,MemStore中的數據是排序的,當MemStore累計到一定閾值時,就會創建一個新的MemStore,并且將老的MemStore添加到flush隊列,由單獨的線程flush到磁盤上,成為一個StoreFile。于此同時, 系統會在zookeeper中記錄一個redo point,表示這個時刻之前的變更已經持久化了(minor compact)。

          StoreFile是只讀的,一旦創建后就不可以再修改。因此Hbase的更新其實是不斷追加的操作。當一個Store中的StoreFile達到一定的閾值后,就會進行一次合并(major compact),將對同一個key的修改合并到一起,形成一個大的StoreFile,當StoreFile的大小達到一定閾值后,又會對 StoreFile進行分割(split),等分為兩個StoreFile。

          由于對表的更新是不斷追加的,處理讀請求時,需要訪問Store中全部的StoreFile和MemStore,將它們按照row key進行合并,由于StoreFile和MemStore都是經過排序的,并且StoreFile帶有內存中索引,通常合并過程還是比較快的。

          實際應用中,可以考慮必要時手動進行major compact,將同一個row key的修改進行合并形成一個大的StoreFile。同時,可以將StoreFile設置大些,減少split的發生。

          2. 寫表操作
          2.1 多HTable并發寫
          創建多個HTable客戶端用于寫操作,提高寫數據的吞吐量,一個例子:


          staticfinalConfiguration conf = HBaseConfiguration.create();
          staticfinalString table_log_name = “user_log”;
          wTableLog =newHTable[tableN];
          for(inti =0; i < tableN; i++) {
          wTableLog[i] =newHTable(conf, table_log_name);
          wTableLog[i].setWriteBufferSize(5*1024*1024);//5MB
          wTableLog[i].setAutoFlush(false);
          }
          2.2 HTable參數設置
          2.2.1 Auto Flush
          通過調用HTable.setAutoFlush(false)方法可以將HTable寫客戶端的自動flush關閉,這樣可以批量寫入數據到HBase,而不是有一條put就執行一次更新,只有當put填滿客戶端寫緩存時,才實際向HBase服務端發起寫請求。默認情況下auto flush是開啟的。

          2.2.2 Write Buffer
          通過調用HTable.setWriteBufferSize(writeBufferSize)方法可以設置HTable客戶端的寫buffer大小,如果新設置的buffer小于當前寫buffer中的數據時,buffer將會被flush到服務端。其中,writeBufferSize的單位是byte字節數,可以根據實際寫入數據量的多少來設置該值。

          2.2.3 WAL Flag
          在HBae中,客戶端向集群中的RegionServer提交數據時(Put/Delete操作),首先會先寫WAL(Write Ahead Log)日志(即HLog,一個RegionServer上的所有Region共享一個HLog),只有當WAL日志寫成功后,再接著寫MemStore,然后客戶端被通知提交數據成功;如果寫WAL日志失敗,客戶端則被通知提交失敗。這樣做的好處是可以做到RegionServer宕機后的數據恢復。

          因此,對于相對不太重要的數據,可以在Put/Delete操作時,通過調用Put.setWriteToWAL(false)或Delete.setWriteToWAL(false)函數,放棄寫WAL日志,從而提高數據寫入的性能。

          值得注意的是:謹慎選擇關閉WAL日志,因為這樣的話,一旦RegionServer宕機,Put/Delete的數據將會無法根據WAL日志進行恢復。

          2.3 批量寫
          通過調用HTable.put(Put)方法可以將一個指定的row key記錄寫入HBase,同樣HBase提供了另一個方法:通過調用HTable.put(List<Put>)方法可以將指定的row key列表,批量寫入多行記錄,這樣做的好處是批量執行,只需要一次網絡I/O開銷,這對于對數據實時性要求高,網絡傳輸RTT高的情景下可能帶來明顯的性能提升。

          2.4 多線程并發寫
          在客戶端開啟多個HTable寫線程,每個寫線程負責一個HTable對象的flush操作,這樣結合定時flush和寫buffer(writeBufferSize),可以既保證在數據量小的時候,數據可以在較短時間內被flush(如1秒內),同時又保證在數據量大的時候,寫buffer一滿就及時進行flush。下面給個具體的例子:


          for(inti =0; i < threadN; i++) {
          Thread th =newThread() {
          publicvoidrun() {
          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并發讀
          創建多個HTable客戶端用于讀操作,提高讀數據的吞吐量,一個例子:

          staticfinalConfiguration conf = HBaseConfiguration.create();
          staticfinalString table_log_name = “user_log”;
          rTableLog =newHTable[tableN];
          for(inti =0; i < tableN; i++) {
          rTableLog[i] =newHTable(conf, table_log_name);
          rTableLog[i].setScannerCaching(50);
          }

          3.2 HTable參數設置
          3.2.1 Scanner Caching
          通過調用HTable.setScannerCaching(int scannerCaching)可以設置HBase scanner一次從服務端抓取的數據條數,默認情況下一次一條。通過將此值設置成一個合理的值,可以減少scan過程中next()的時間開銷,代價是scanner需要通過客戶端的內存來維持這些被cache的行記錄。

          3.2.2 Scan Attribute Selection
          scan時指定需要的Column Family,可以減少網絡傳輸數據量,否則默認scan操作會返回整行所有Column Family的數據。

          3.2.3 Close ResultScanner
          通過scan取完數據后,記得要關閉ResultScanner,否則RegionServer可能會出現問題(對應的Server資源無法釋放)。

          3.3 批量讀
          通過調用HTable.get(Get)方法可以根據一個指定的row key獲取一行記錄,同樣HBase提供了另一個方法:通過調用HTable.get(List)方法可以根據一個指定的row key列表,批量獲取多行記錄,這樣做的好處是批量執行,只需要一次網絡I/O開銷,這對于對數據實時性要求高而且網絡傳輸RTT高的情景下可能帶來明顯的性能提升。

          3.4 多線程并發讀
          在客戶端開啟多個HTable讀線程,每個讀線程負責通過HTable對象進行get操作。下面是一個多線程并發讀取HBase,獲取店鋪一天內各分鐘PV值的例子:

          publicclassDataReaderServer {
          //獲取店鋪一天內各分鐘PV值的入口函數
          publicstaticConcurrentHashMap getUnitMinutePV(longuid,longstartStamp,longendStamp){
          longmin = startStamp;
          intcount = (int)((endStamp - startStamp) / (60*1000));
          List lst =newArrayList();
          for(inti =0; i <= count; i++) {
          min = startStamp + i *60*1000;
          lst.add(uid +"_"+ min);
          }
          returnparallelBatchMinutePV(lst);
          }
          //多線程并發查詢,獲取分鐘PV值
          privatestaticConcurrentHashMap parallelBatchMinutePV(List lstKeys){
          ConcurrentHashMap hashRet =newConcurrentHashMap();
          intparallel =3;
          List<List<String>> lstBatchKeys =null;
          if(lstKeys.size() < parallel ){
          lstBatchKeys =newArrayList<List<String>>(1);
          lstBatchKeys.add(lstKeys);
          }
          else{
          lstBatchKeys =newArrayList<List<String>>(parallel);
          for(inti =0; i < parallel; i++ ){
          List lst =newArrayList();
          lstBatchKeys.add(lst);
          }

          for(inti =0; i < lstKeys.size() ; i ++ ){
          lstBatchKeys.get(i%parallel).add(lstKeys.get(i));
          }
          }

          List >> futures =newArrayList >>(5);

          ThreadFactoryBuilder builder =newThreadFactoryBuilder();
          builder.setNameFormat("ParallelBatchQuery");
          ThreadFactory factory = builder.build();
          ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(lstBatchKeys.size(), factory);

          for(List keys : lstBatchKeys){
          Callable< ConcurrentHashMap > callable =newBatchMinutePVCallable(keys);
          FutureTask< ConcurrentHashMap > future = (FutureTask< ConcurrentHashMap >) executor.submit(callable);
          futures.add(future);
          }
          executor.shutdown();

          // Wait for all the tasks to finish
          try{
          booleanstillRunning = !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();
          }
          }

          returnhashRet;
          }
          //一個線程批量查詢,獲取分鐘PV值
          protectedstaticConcurrentHashMap getBatchMinutePV(List lstKeys){
          ConcurrentHashMap hashRet =null;
          List lstGet =newArrayList();
          String[] splitValue =null;
          for(String s : lstKeys) {
          splitValue = s.split("_");
          longuid = Long.parseLong(splitValue[0]);
          longmin = Long.parseLong(splitValue[1]);
          byte[] key =newbyte[16];
          Bytes.putLong(key,0, uid);
          Bytes.putLong(key,8, min);
          Get g =newGet(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 =newConcurrentHashMap(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());
          }
          }
          }
          }

          returnhashRet;
          }
          }
          //調用接口類,實現Callable接口
          classBatchMinutePVCallableimplementsCallable>{
          privateList keys;

          publicBatchMinutePVCallable(List lstKeys ) {
          this.keys = lstKeys;
          }

          publicConcurrentHashMap call()throwsException {
          returnDataReadServer.getBatchMinutePV(keys);
          }
          }

          3.5 緩存查詢結果
          對于頻繁查詢HBase的應用場景,可以考慮在應用程序中做緩存,當有新的查詢請求時,首先在緩存中查找,如果存在則直接返回,不再查詢HBase;否則對HBase發起讀請求查詢,然后在應用程序中將查詢結果緩存起來。至于緩存的替換策略,可以考慮LRU等常用的策略。

          3.6 Blockcache
          HBase上Regionserver的內存分為兩個部分,一部分作為Memstore,主要用來寫;另外一部分作為BlockCache,主要用于讀。

          寫請求會先寫入Memstore,Regionserver會給每個region提供一個Memstore,當Memstore滿64MB以后,會啟動 flush刷新到磁盤。當Memstore的總大小超過限制時(heapsize * hbase.regionserver.global.memstore.upperLimit * 0.9),會強行啟動flush進程,從最大的Memstore開始flush直到低于限制。

          讀請求先到Memstore中查數據,查不到就到BlockCache中查,再查不到就會到磁盤上讀,并把讀的結果放入BlockCache。由于BlockCache采用的是LRU策略,因此BlockCache達到上限(heapsize * hfile.block.cache.size * 0.85)后,會啟動淘汰機制,淘汰掉最老的一批數據。

          一個Regionserver上有一個BlockCache和N個Memstore,它們的大小之和不能大于等于heapsize * 0.8,否則HBase不能啟動。默認BlockCache為0.2,而Memstore為0.4。對于注重讀響應時間的系統,可以將 BlockCache設大些,比如設置BlockCache=0.4,Memstore=0.39,以加大緩存的命中率。

          有關BlockCache機制,請參考這里:HBase的Block cache,HBase的blockcache機制,hbase中的緩存的計算與使用。

          4.數據計算
          4.1 服務端計算
          Coprocessor運行于HBase RegionServer服務端,各個Regions保持對與其相關的coprocessor實現類的引用,coprocessor類可以通過RegionServer上classpath中的本地jar或HDFS的classloader進行加載。

          目前,已提供有幾種coprocessor:

          Coprocessor:提供對于region管理的鉤子,例如region的open/close/split/flush/compact等;
          RegionObserver:提供用于從客戶端監控表相關操作的鉤子,例如表的get/put/scan/delete等;
          Endpoint:提供可以在region上執行任意函數的命令觸發器。一個使用例子是RegionServer端的列聚合,這里有代碼示例。
          以上只是有關coprocessor的一些基本介紹,本人沒有對其實際使用的經驗,對它的可用性和性能數據不得而知。感興趣的同學可以嘗試一下,歡迎討論。

          4.2 寫端計算
          4.2.1 計數
          HBase本身可以看作是一個可以水平擴展的Key-Value存儲系統,但是其本身的計算能力有限(Coprocessor可以提供一定的服務端計算),因此,使用HBase時,往往需要從寫端或者讀端進行計算,然后將最終的計算結果返回給調用者。舉兩個簡單的例子:

          PV計算:通過在HBase寫端內存中,累加計數,維護PV值的更新,同時為了做到持久化,定期(如1秒)將PV計算結果同步到HBase中,這樣查詢端最多會有1秒鐘的延遲,能看到秒級延遲的PV結果。
          分鐘PV計算:與上面提到的PV計算方法相結合,每分鐘將當前的累計PV值,按照rowkey + minute作為新的rowkey寫入HBase中,然后在查詢端通過scan得到當天各個分鐘以前的累計PV值,然后順次將前后兩分鐘的累計PV值相減,就得到了當前一分鐘內的PV值,從而最終也就得到當天各個分鐘內的PV值。

          4.2.2 去重
          對于UV的計算,就是個去重計算的例子。分兩種情況:

          如果內存可以容納,那么可以在Hash表中維護所有已經存在的UV標識,每當新來一個標識時,通過快速查找Hash確定是否是一個新的UV,若是則UV值加1,否則UV值不變。另外,為了做到持久化或提供給查詢接口使用,可以定期(如1秒)將UV計算結果同步到HBase中。
          如果內存不能容納,可以考慮采用Bloom Filter來實現,從而盡可能的減少內存的占用情況。除了UV的計算外,判斷URL是否存在也是個典型的應用場景。

          4.3 讀端計算
          如果對于響應時間要求比較苛刻的情況(如單次http請求要在毫秒級時間內返回),個人覺得讀端不宜做過多復雜的計算邏輯,盡量做到讀端功能單一化:即從HBase RegionServer讀到數據(scan或get方式)后,按照數據格式進行簡單的拼接,直接返回給前端使用。當然,如果對于響應時間要求一般,或者業務特點需要,也可以在讀端進行一些計算邏輯。

          5.總結
          作為一個Key-Value存儲系統,HBase并不是萬能的,它有自己獨特的地方。因此,基于它來做應用時,我們往往需要從多方面進行優化改進(表設計、讀表操作、寫表操作、數據計算等),有時甚至還需要從系統級對HBase進行配置調優,更甚至可以對HBase本身進行優化。這屬于不同的層次范疇。

          總之,概括來講,對系統進行優化時,首先定位到影響你的程序運行性能的瓶頸之處,然后有的放矢進行針對行的優化。如果優化后滿足你的期望,那么就可以停止優化;否則繼續尋找新的瓶頸之處,開始新的優化,直到滿足性能要求。

          以上就是從項目開發中總結的一點經驗,如有不對之處,歡迎大家不吝賜教。 

          posted on 2012-11-29 21:43 paulwong 閱讀(6756) 評論(0)  編輯  收藏 所屬分類: HBASE

          主站蜘蛛池模板: 陕西省| 渝中区| 股票| 甘谷县| 南开区| 堆龙德庆县| 二连浩特市| 台南市| 高邑县| 上饶县| 高阳县| 宝应县| 华蓥市| 尼勒克县| 隆尧县| 天台县| 大余县| 新邵县| 稻城县| 铜鼓县| 柳州市| 文登市| 松滋市| 信丰县| 法库县| 鲜城| 股票| 年辖:市辖区| 乌鲁木齐市| 伊通| 铜梁县| 霍州市| 荔波县| 海宁市| 安西县| 大丰市| 怀集县| 沾益县| 酉阳| 奈曼旗| 嘉兴市|