HBase協處理器實戰

主要內容：

1. HBase協處理器介紹

2. 觀察者（Observer）

3. 終端（endpoint）

-------------------------------------------------------------------------------------------------------------------------------------------------------------------

1. HBase協處理器介紹

     系統協處理器能夠全局導入region server上的全部數據表，表協處理器便是用戶能夠指定一張表使用協處理器。Hbase協處理器（Coprocessor）有兩種類型：Observer Coprocessors 和Endpoint Coprocessor。

     前者相似觸發器，在特定的事件發生時候觸發，後者相似存儲過程，執行數據計算。觀察者協處理器在不少地方可能用到這些，好比：數據安全權限限制，數據外鍵參考或者一致性，二級索引，主要類型有：RegionObserver，RegionServerObserver，MasterObserver，WalObserver。

2. 觀察者（Observer）

      觀察者的設計意圖是容許用戶經過插入代碼來重載協處理器框架的upcall方法，而具體的事件觸發的callback方法由HBase的核心代碼來執行。協處理器框架處理全部的callback調用細節，協處理器自身只須要插入添加或者改變的功能。

     以HBase0.92版本爲例，它提供了三種觀察者接口：

• RegionObserver：提供客戶端的數據操縱事件鉤子：Get、Put、Delete、Scan等。
• WALObserver：提供WAL相關操做鉤子。
• MasterObserver：提供DDL類型的操做鉤子。如建立、刪除、修改數據表等。

     這些接口能夠同時使用在同一個地方，按照不一樣優先級順序執行.用戶能夠任意基於協處理器實現複雜的HBase功能層。HBase有不少種事件能夠觸發觀察者方法，這些事件與方法從HBase0.92版本起，都會集成在HBase API中。不過這些API可能會因爲各類緣由有所改動，不一樣版本的接口改動比較大，具體參考Java Doc，RegionObserver工做原理以下圖所示。

圖1 RegionObserver工做原理

 

3. 終端（endpoint）

       HBase 提供了客戶端 Java 包 org.apache.hadoop.hbase.client.coprocessor。它提供如下三種方法來調用協處理器提供的服務：

• Table.coprocessorService(byte[])
• Table.coprocessorService(Class, byte[], byte[],Batch.Call),
• Table.coprocessorService(Class, byte[], byte[], Batch.Call, Batch.Callback)

      Endpoint 協處理器在Region上下文中運行，一個 HBase 表可能有多個Region。所以客戶端能夠指定調用某一個單個Region上的協處理器，在單個Region上進行處理並返回必定結果；也能夠調用必定範圍內的若干Region上的協處理器併發執行，並對結果進行彙總處理。針對不一樣的須要，能夠選擇如下三種方法。

    （1）調用單個Region上的協處理器RPC

     第一個方法使用API coprocessorService(byte[])，這個函數只調用單個Region上的協處理器。

     該方法採用RowKey指定Region。這是由於HBase的客戶端不多會直接操做Region，通常不須要知道 Region 的名字；何況在 HBase 中，Region 名會隨時改變，因此用rowkey來指定Region是最合理的方式。使用 rowkey 能夠指定惟一的一個Region，若是給定的 rowkey 並不存在，只要在某個Region的rowkey範圍內，依然能夠用來指定該Region。好比Region1處理[row1, row100]這個區間內的數據，則 rowkey=row1 就由Region1來負責處理，換句話說，咱們能夠用row1來指定Region1，不管 rowkey 等於」row1」的記錄是否存在。

 圖2 調用單個Region上的協處理器

      coprocessorService 方法返回類型爲 CoprocessorRpcChannel 的對象，該 RPC 通道鏈接到由 rowkey 指定的 Region 上，經過這個通道，就能夠調用該 Region 上部署的協處理器 RPC。咱們已經經過 Protobuf 定義了 RPC Service。調用 Service 的 newBlockingStub() 方法，將 CoprocessorRpcChannel 做爲輸入參數，就能夠獲得 RPC 調用的 stub 對象，進而調用遠端的 RPC。

代碼1 獲取單個Region的rowcount

 1 long singleRegionCount(String tableName, String rowkey,boolean reCount)
 2 {
 3  long rowcount = 0;
 4  try{
 5  Configuration config = new Configuration();
 6  HConnection conn = HConnectionManager.createConnection(config);
 7  HTableInterface tbl = conn.getTable(tableName);
 8  //獲取 Channel
 9  CoprocessorRpcChannel channel = tbl.coprocessorService(rowkey.getBytes());
10 org.ibm.developerworks.getRowCount.ibmDeveloperWorksService.BlockingInterface service =
11 org.ibm.developerworks.getRowCount.ibmDeveloperWorksService.newBlockingStub(channel);
12   //設置 RPC 入口參數
13 org.ibm.developerworks.getRowCount.getRowCountRequest.Builder request = 
14 org.ibm.developerworks.getRowCount.getRowCountRequest.newBuilder();
15  request.setReCount(reCount);
16   //調用 RPC
17  org.ibm.developerworks.getRowCount.getRowCountResponse ret =
18  service.getRowCount(null, request.build());
19  
20  //解析結果
21  rowcount = ret.getRowCount();
22  }
23  catch(Exception e) {e.printStackTrace();}
24  return rowcount;
25  }

     （2）調用多個 Region 上的協處理器 RPC，不使用 callback

      有時候客戶端須要調用多個Region上的同一個協處理器，好比須要統計整個table的rowcount，在這種狀況下，須要全部的Region都參與進來，分別統計本身Region內部的rowcount並返回客戶端，最終客戶端將全部 Region 的返回結果彙總，就能夠獲得整張表的 rowcount。

      這意味着該客戶端同時和多個 Region 進行批處理交互。具體方法是，收集每一個Region的startkey，而後循環調用第一種coprocessorService方法：用每個Region的startkey做爲入口參數，得到RPC通道，建立 stub對象，進而逐一調用每一個Region上的協處理器RPC。這種作法須要寫不少的代碼，爲此HBase提供了兩種更加簡單的coprocessorService方法來處理多個Region的協處理器調用。先來看第一種方法 coprocessorService(Class, byte[],byte[],Batch.Call)，該方法有 4 個入口參數。第一個參數是實現RPC的Service類，即前文中的ibmDeveloperWorksService類。經過它，HBase 就能夠找到相應的部署在Region上的協處理器，一個Region上能夠部署多個協處理器，客戶端必須經過指定Service 類來區分究竟須要調用哪一個協處理器提供的服務。

      要調用哪些 Region 上的服務則由startkey和endkey來肯定，經過rowkey範圍便可肯定多個Region。爲此，coprocessorService 方法的第二個和第三個參數分別是startkey和endkey，凡是落在[startkey，endkey] 區間內的Region都會參與本次調用。

      第四個參數是接口類Batch.Call。它定義瞭如何調用協處理器，用戶經過重載該接口的call()方法來實現客戶端的邏輯。在call()方法內，能夠調用RPC，並對返回值進行任意處理。即前文代碼1中所作的事情。coprocessorService將負責對每一個Region調用這個call方法。

      coprocessorService 方法的返回值是一個map類型的集合。該集合的key是Region名字，value是Batch.Call.call方法的返回值。該集合能夠看做是全部Region的協處理器RPC返回的結果集。客戶端代碼能夠遍歷該集合對全部的結果進行彙總處理。

     這種coprocessorService方法的大致工做流程以下。首先它分析startkey和endkey，找到該區間內的全部Region，假設存放在regionList 中。而後，遍歷regionList，爲每個Region調用Batch.Call，在該接口內，用戶定義了具體的RPC調用邏輯。最後coprocessorService將全部Batch.Call.call()的返回值加入結果集合並返回。以下圖所示：

 

 

圖3 調用多個Region上的協處理器——不使用callback

       （3）調用多個 Region 上的協處理器 RPC，使用 callback

      coprocessorService 的第三種方法比第二個方法多了一個參數callback。coprocessorService 第二個方法內部使用HBase自帶的缺省callback，該缺省callback將每一個Region的返回結果都添加到一個map類型的結果集中，並將該集合做爲coprocessorService方法的返回值。

      這個結果集合的key是Region名字，value是call方法的返回值。採用這種方法，客戶端代碼須要將RPC執行結果先保存在一個集合中，再進入一個循環，遍歷結果集合進一步處理。有些狀況下這種使用集合的開銷是沒必要要的。對每一個 Region 的返回結果直接進行處理能夠省去這些開銷。具體過程以下圖所示：

圖4 調用多個Region上的協處理器——使用callback

      HBase 提供第三種 coprocessorService 方法容許用戶定義 callback 行爲，coprocessorService 會爲每個 RPC 返回結果調用該 callback，用戶能夠在 callback 中執行須要的邏輯，好比執行 sum 累加。用第二種方法的狀況下，每一個 Region 協處理器 RPC 的返回結果先放入一個列表，全部的 Region 都返回後，用戶代碼再從該列表中取出每個結果進行累加；用第三種方法，直接在 callback 中進行累加，省掉了建立結果集合和遍歷該集合的開銷，效率會更高一些。所以咱們只須要額外定義一個 callback 便可，callback 是一個 Batch.Callback 接口類，用戶須要重載其 update 方法。