碼上快樂

相關內容    簡體    繁體

Hbase開發實例

本文轉載自   查看原文   2016-10-24 01:25   2149 

1、開發環境

在進行Hbase開發前，需要安裝JDK、Hadoop和HBase

根據自己的安裝環境修改版本信息，使用Maven構建項目，在pom.xml中添加hbase的依賴

2、初始化配置

設置HBase的配置，如ZooKeeper的地址、端口號等等。可以通過org.apache.hadoop.conf.Configuration.set方法手工設置HBase的配置信息，也可以直接將HBase的hbase-site.xml配置文件引入項目即可。

3、常見API的使用

HBase的常用操作包括建表、插入表數據、刪除表數據、獲取一行數據、表掃描、刪除列族、刪除表等等，下面給出具體代碼。

3.1 創建數據庫表

  1.  // 創建數據庫表  
  2. public static void createTable(String tableName, String[] columnFamilys) throws IOException {  
  3.     // 建立一個數據庫的連接  
  4.     Connection conn = ConnectionFactory.createConnection(conf);  
  5.     // 創建一個數據庫管理員  
  6.     HBaseAdmin hAdmin = (HBaseAdmin) conn.getAdmin();  
  7.     if (hAdmin.tableExists(tableName)) {  
  8.         System.out.println(tableName + "表已存在");  
  9.         conn.close();  
  10.         System.exit(0);  
  11.     } else {  
  12.         // 新建一個表描述  
  13.         HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));  
  14.         // 在表描述里添加列族  
  15.         for (String columnFamily : columnFamilys) {  
  16.             tableDesc.addFamily(new HColumnDescriptor(columnFamily));  
  17.         }  
  18.         // 根據配置好的表描述建表  
  19.         hAdmin.createTable(tableDesc);  
  20.         System.out.println("創建" + tableName + "表成功");  
  21.     }  
  22.     conn.close();  
  23. }  

3.2 添加一條數據

  1.  // 添加一條數據  
  2. public static void addRow(String tableName, String rowKey, String columnFamily, String column, String value)   
  3.         throws IOException {  
  4.     // 建立一個數據庫的連接  
  5.     Connection conn = ConnectionFactory.createConnection(conf);  
  6.     // 獲取表  
  7.     HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));  
  8.     // 通過rowkey創建一個put對象  
  9.     Put put = new Put(Bytes.toBytes(rowKey));  
  10.     // 在put對象中設置列族、列、值  
  11.     put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));  
  12.     // 插入數據,可通過put(List<Put>)批量插入  
  13.     table.put(put);  
  14.     // 關閉資源  
  15.     table.close();  
  16.     conn.close();  
  17. }  

3.3 獲取一條數

  1. // 通過rowkey獲取一條數據  
  2. public static void getRow(String tableName, String rowKey) throws IOException {  
  3.     // 建立一個數據庫的連接  
  4.     Connection conn = ConnectionFactory.createConnection(conf);  
  5.     // 獲取表  
  6.     HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));  
  7.     // 通過rowkey創建一個get對象  
  8.     Get get = new Get(Bytes.toBytes(rowKey));  
  9.     // 輸出結果  
  10.     Result result = table.get(get);  
  11.     for (Cell cell : result.rawCells()) {  
  12.         System.out.println(  
  13.                 "行鍵:" + new String(CellUtil.cloneRow(cell)) + "\t" +  
  14.                 "列族:" + new String(CellUtil.cloneFamily(cell)) + "\t" +   
  15.                 "列名:" + new String(CellUtil.cloneQualifier(cell)) + "\t" +   
  16.                 "值:" + new String(CellUtil.cloneValue(cell)) + "\t" +  
  17.                 "時間戳:" + cell.getTimestamp());  
  18.     }  
  19.     // 關閉資源  
  20.     table.close();  
  21.     conn.close();  
  22. }  ​

3.4 全表掃描

  1. // 全表掃描  
  2.     public static void scanTable(String tableName) throws IOException {  
  3.         // 建立一個數據庫的連接  
  4.         Connection conn = ConnectionFactory.createConnection(conf);  
  5.         // 獲取表  
  6.         HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));  
  7.         // 創建一個掃描對象  
  8.         Scan scan = new Scan();  
  9.         // 掃描全表輸出結果  
  10.         ResultScanner results = table.getScanner(scan);  
  11.         for (Result result : results) {  
  12.             for (Cell cell : result.rawCells()) {  
  13.                 System.out.println(  
  14.                         "行鍵:" + new String(CellUtil.cloneRow(cell)) + "\t" +  
  15.                         "列族:" + new String(CellUtil.cloneFamily(cell)) + "\t" +   
  16.                         "列名:" + new String(CellUtil.cloneQualifier(cell)) + "\t" +   
  17.                         "值:" + new String(CellUtil.cloneValue(cell)) + "\t" +  
  18.                         "時間戳:" + cell.getTimestamp());  
  19.             }  
  20.         }  
  21.         // 關閉資源  
  22.         results.close();  
  23.         table.close();  
  24.         conn.close();  
  25. }  

3.5 刪除一條數據

  1. // 刪除一條數據  
  2. public static void delRow(String tableName, String rowKey) throws IOException {  
  3.     // 建立一個數據庫的連接  
  4.     Connection conn = ConnectionFactory.createConnection(conf);  
  5.     // 獲取表  
  6.     HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));  
  7.     // 刪除數據  
  8.     Delete delete = new Delete(Bytes.toBytes(rowKey));  
  9.     table.delete(delete);  
  10.     // 關閉資源  
  11.     table.close();  
  12.     conn.close();  
  13. }  
  14. ​

3.6 刪除多條數據

  1. // 刪除多條數據  
  2. public static void delRows(String tableName, String[] rows) throws IOException {  
  3.     // 建立一個數據庫的連接  
  4.     Connection conn = ConnectionFactory.createConnection(conf);  
  5.     // 獲取表  
  6.     HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));  
  7.     // 刪除多條數據  
  8.     List<Delete> list = new ArrayList<Delete>();  
  9.     for (String row : rows) {  
  10.         Delete delete = new Delete(Bytes.toBytes(row));  
  11.         list.add(delete);  
  12.     }  
  13.     table.delete(list);  
  14.     // 關閉資源  
  15.     table.close();  
  16.     conn.close();  
  17. }  

3.7 刪除列族

  1. // 刪除列族  
  2. public static void delColumnFamily(String tableName, String columnFamily) throws IOException {  
  3.     // 建立一個數據庫的連接  
  4.     Connection conn = ConnectionFactory.createConnection(conf);  
  5.     // 創建一個數據庫管理員  
  6.     HBaseAdmin hAdmin = (HBaseAdmin) conn.getAdmin();  
  7.     // 刪除一個表的指定列族  
  8.     hAdmin.deleteColumn(tableName, columnFamily);  
  9.     // 關閉資源  
  10.     conn.close();  
  11. }  

3.8 刪除數據庫表

  1. // 刪除數據庫表  
  2. public static void deleteTable(String tableName) throws IOException {  
  3.     // 建立一個數據庫的連接  
  4.     Connection conn = ConnectionFactory.createConnection(conf);  
  5.     // 創建一個數據庫管理員  
  6.     HBaseAdmin hAdmin = (HBaseAdmin) conn.getAdmin();  
  7.     if (hAdmin.tableExists(tableName)) {  
  8.         // 失效表  
  9.         hAdmin.disableTable(tableName);  
  10.         // 刪除表  
  11.         hAdmin.deleteTable(tableName);  
  12.         System.out.println("刪除" + tableName + "表成功");  
  13.         conn.close();  
  14.     } else {  
  15.         System.out.println("需要刪除的" + tableName + "表不存在");  
  16.         conn.close();  
  17.         System.exit(0);  
  18.     }  
  19. }  

3.9 追加插入

  1. // 追加插入(將原有value的后面追加新的value，如原有value=a追加value=bc則最后的value=abc)  
  2. public static void appendData(String tableName, String rowKey, String columnFamily, String column, String value)   
  3.         throws IOException {  
  4.     // 建立一個數據庫的連接  
  5.     Connection conn = ConnectionFactory.createConnection(conf);  
  6.     // 獲取表  
  7.     HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));  
  8.     // 通過rowkey創建一個append對象  
  9.     Append append = new Append(Bytes.toBytes(rowKey));  
  10.     // 在append對象中設置列族、列、值  
  11.     append.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));  
  12.     // 追加數據  
  13.     table.append(append);  
  14.     // 關閉資源  
  15.     table.close();  
  16.     conn.close();  
  17. }  
  18. ​

3.10 符合條件后添加數據

  1. // 符合條件后添加數據(只能針對某一個rowkey進行原子操作)  
  2. public static boolean checkAndPut(String tableName, String rowKey, String columnFamilyCheck, String columnCheck, String valueCheck, String columnFamily, String column, String value) throws IOException {  
  3.     // 建立一個數據庫的連接  
  4.     Connection conn = ConnectionFactory.createConnection(conf);  
  5.     // 獲取表  
  6.     HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));  
  7.     // 設置需要添加的數據  
  8.     Put put = new Put(Bytes.toBytes(rowKey));  
  9.     put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));  
  10.     // 當判斷條件為真時添加數據  
  11.     boolean result = table.checkAndPut(Bytes.toBytes(rowKey), Bytes.toBytes(columnFamilyCheck),   
  12.             Bytes.toBytes(columnCheck), Bytes.toBytes(valueCheck), put);  
  13.     // 關閉資源  
  14.     table.close();  
  15.     conn.close();  
  16.       
  17.     return result;  
  18. }  
  19. ​

3.11 符合條件后刪除數據

  1. // 符合條件后刪除數據(只能針對某一個rowkey進行原子操作)  
  2. public static boolean checkAndDelete(String tableName, String rowKey, String columnFamilyCheck, String columnCheck,   
  3.         String valueCheck, String columnFamily, String column) throws IOException {  
  4.     // 建立一個數據庫的連接  
  5.     Connection conn = ConnectionFactory.createConnection(conf);  
  6.     // 獲取表  
  7.     HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));  
  8.     // 設置需要刪除的delete對象  
  9.     Delete delete = new Delete(Bytes.toBytes(rowKey));  
  10.     delete.addColumn(Bytes.toBytes(columnFamilyCheck), Bytes.toBytes(columnCheck));  
  11.     // 當判斷條件為真時添加數據  
  12.     boolean result = table.checkAndDelete(Bytes.toBytes(rowKey), Bytes.toBytes(columnFamilyCheck), Bytes.toBytes(columnCheck),   
  13.             Bytes.toBytes(valueCheck), delete);  
  14.     // 關閉資源  
  15.     table.close();  
  16.     conn.close();  
  17.   
  18.     return result;  
  19. }  

3.12 計數器

  1. // 計數器(amount為正數則計數器加，為負數則計數器減，為0則獲取當前計數器的值)  
  2. public static long incrementColumnValue(String tableName, String rowKey, String columnFamily, String column, long amount)   
  3.         throws IOException {  
  4.     // 建立一個數據庫的連接  
  5.     Connection conn = ConnectionFactory.createConnection(conf);  
  6.     // 獲取表  
  7.     HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));  
  8.     // 計數器  
  9.     long result = table.incrementColumnValue(Bytes.toBytes(rowKey), Bytes.toBytes(columnFamily), Bytes.toBytes(column), amount);  
  10.     // 關閉資源  
  11.     table.close();  
  12.     conn.close();  
  13.       
  14.     return result;  
  15. }  

4、內置過濾器的使用

HBase為篩選數據提供了一組過濾器，通過這個過濾器可以在HBase中數據的多個維度（行、列、數據版本）上進行對數據的篩選操作，也就是說過濾器最終能夠篩選的數據能夠細化到具體的一個存儲單元格上（由行鍵、列名、時間戳定位）。通常來說，通過行鍵、值來篩選數據的應用場景較多。需要說明的是，過濾器會極大地影響查詢效率。所以，在數據量較大的數據表中，應盡量避免使用過濾器。

下面介紹一些常用的HBase內置過濾器的用法：

1、RowFilter：篩選出匹配的所有的行。使用BinaryComparator可以篩選出具有某個行鍵的行，或者通過改變比較運算符（下面的例子中是CompareFilter.CompareOp.EQUAL）來篩選出符合某一條件的多條數據，如下示例就是篩選出行鍵為row1的一行數據。

  1. // 篩選出匹配的所有的行  
  2. Filter rf = new RowFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("row1")));

2、PrefixFilter：篩選出具有特定前綴的行鍵的數據。這個過濾器所實現的功能其實也可以由RowFilter結合RegexComparator來實現，不過這里提供了一種簡便的使用方法，如下示例就是篩選出行鍵以row為前綴的所有的行。

  1. // 篩選匹配行鍵的前綴成功的行  
  2. Filter pf = new PrefixFilter(Bytes.toBytes("row"));  

3、KeyOnlyFilter：這個過濾器唯一的功能就是只返回每行的行鍵，值全部為空，這對於只關注於行鍵的應用場景來說非常合適，這樣忽略掉其值就可以減少傳遞到客戶端的數據量，能起到一定的優化作用。

  1. // 返回所有的行鍵，但值全是空  
  2. Filter kof = new KeyOnlyFilter();

4、RandomRowFilter：按照一定的幾率（<=0會過濾掉所有的行，>=1會包含所有的行）來返回隨機的結果集，對於同樣的數據集，多次使用同一個RandomRowFilter會返回不同的結果集，對於需要隨機抽取一部分數據的應用場景，可以使用此過濾器。

  1. // 隨機選出一部分的行  
  2. Filter rrf = new RandomRowFilter((float) 0.8);

5、InclusiveStopFilter：掃描的時候，我們可以設置一個開始行鍵和一個終止行鍵，默認情況下，這個行鍵的返回是前閉后開區間，即包含起始行，但不包含終止行。如果我們想要同時包含起始行和終止行，那么可以使用此過濾器。

  1. // 包含了掃描的上限在結果之內  
  2. Filter isf = new InclusiveStopFilter(Bytes.toBytes("row1"));

6、FirstKeyOnlyFilter：如果想要返回的結果集中只包含第一列的數據，那么這個過濾器能夠滿足要求。它在找到每行的第一列之后會停止掃描，從而使掃描的性能也得到了一定的提升。

  1. // 篩選出每行的第一個單元格  
  2. Filter fkof = new FirstKeyOnlyFilter();

7、ColumnPrefixFilter：它按照列名的前綴來篩選單元格，如果我們想要對返回的列的前綴加以限制的話，可以使用這個過濾器。

  1. // 篩選出前綴匹配的列  
  2. Filter cpf = new ColumnPrefixFilter(Bytes.toBytes("qual1"));

8、ValueFilter：按照具體的值來篩選單元格的過濾器，這會把一行中值不能滿足的單元格過濾掉，如下面的構造器，對於每一行的一個列，如果其對應的值不包含ROW2_QUAL1，那么這個列就不會返回給客戶端。

  1. // 篩選某個（值的條件滿足的）特定的單元格  
  2. Filter vf = new ValueFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator("ROW2_QUAL1"));

9、ColumnCountGetFilter：這個過濾器在遇到一行的列數超過我們所設置的限制值的時候，結束掃描操作。

  1. // 如果突然發現一行中的列數超過設定的最大值時，整個掃描操作會停止  
  2. Filter ccf = new ColumnCountGetFilter(2);

10、SingleColumnValueFilter：用一列的值決定這一行的數據是否被過濾，可對它的對象調用setFilterIfMissing方法，默認的參數是false。其作用是，對於咱們要使用作為條件的列，如果參數為true，這樣的行將會被過濾掉，如果參數為false，這樣的行會包含在結果集中。

  1. // 將滿足條件的列所在的行過濾掉  
  2.  
  3. SingleColumnValueFilter scvf = new SingleColumnValueFilter(    
  4.          Bytes.toBytes("colfam1"),     
  5.          Bytes.toBytes("qual2"),     
  6.          CompareFilter.CompareOp.NOT_EQUAL,     
  7.          new SubstringComparator("BOGUS"));    
  8. scvf.setFilterIfMissing(true);  
  9. ​

11、SingleColumnValueExcludeFilter：這個過濾器與第10種過濾器唯一的區別就是，作為篩選條件的列，其行不會包含在返回的結果中。

12、SkipFilter：這是一種附加過濾器，其與ValueFilter結合使用，如果發現一行中的某一列不符合條件，那么整行就會被過濾掉。

  1. // 發現某一行中的一列需要過濾時，整個行就會被過濾掉  
  2. Filter skf = new SkipFilter(vf);

13、WhileMatchFilter：使用這個過濾器，當遇到不符合設定條件的數據的時候，整個掃描結束。

  1. // 當遇到不符合過濾器rf設置的條件時，整個掃描結束  
  2. Filter wmf = new WhileMatchFilter(rf);

14. FilterList：可以用於綜合使用多個過濾器。其有兩種關系： Operator.MUST_PASS_ONE表示關系AND，Operator.MUST_PASS_ALL表示關系OR，並且FilterList可以嵌套使用，使得我們能夠表達更多的需求。

  1. // 綜合使用多個過濾器，AND和OR兩種關系  
  2. List<Filter> filters = new ArrayList<Filter>();    
  3. filters.add(rf);    
  4. filters.add(vf);    
  5. FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ALL,filters);

5、HBase與MapReduce

我們知道，在偽分布式模式和完全分布式模式下的HBase是架構在HDFS之上的，因此完全可以將MapReduce編程框架和HBase結合起來使用。也就是說，將HBase作為底層存儲結構，MapReduce調用HBase進行特殊的處理，這樣能夠充分結合HBase分布式大型數據庫和MapReduce並行計算的優點。

HBase實現了TableInputFormatBase類，該類提供了對表數據的大部分操作，其子類TableInputFormat則提供了完整的實現，用於處理表數據並生成鍵值對。TableInputFormat類將數據表按照Region分割成split，即有多少個Regions就有多個splits，然后將Region按行鍵分成<key,value>對，key值對應與行鍵，value值為該行所包含的數據。

HBase實現了MapReduce計算框架對應的TableMapper類和TableReducer類。其中，TableMapper類並沒有具體的功能，只是將輸入的<key,value>對的類型分別限定為Result和ImmutableBytesWritable。IdentityTableMapper類和IdentityTableReducer類則是上述兩個類的具體實現，其和Mapper類和Reducer類一樣，只是簡單地將<key,value>對輸出到下一個階段。

HBase實現的TableOutputFormat將輸出的<key,value>對寫到指定的HBase表中，該類不會對WAL（Write-Ahead Log）進行操作，即如果服務器發生故障將面臨丟失數據的風險。可以使用MultipleTableOutputFormat類解決這個問題，該類可以對是否寫入WAL進行設置。

為了能使Hadoop集群上運行HBase程序，還需要把相關的類文件引入Hadoop集群上，不然會出現ClassNotFoundException錯誤。其具體方法是可在hadoop的環境配置文件hadoop-env.sh中引入HBASE_HOME和HBase的相關jar包，或者直接將HBase的jar包打包到應用程序文件中。

下例子是將MapReduce和HBase結合起來的WordCount程序，它首先從指定文件中搜集數據，進行統計計算，最后將結果存儲到HBase中。代碼省略。

6、HBase的Bulkload

HBase可以讓我們隨機的、實時的訪問大數據，但是怎樣有效的將數據導入到HBase呢？HBase有多種導入數據的方法，最直接的方法就是在MapReduce作業中使用TableOutputFormat作為輸出，或者使用標准的客戶端API，但是這些都不是非常有效的方法。

如果HDFS中有海量數據要導入HBase，可以先將這些數據生成HFile文件，然后批量導入HBase的數據表中，這樣可以極大地提升數據導入HBase的效率。這就是HBase的Bulkload，即利用MapReduce作業輸出HBase內部數據格式的表數據，然后將生成的StoreFiles直接導入到集群中。與使用HBase API相比，使用Bulkload導入數據占用更少的CPU和網絡資源。兩個表之間的數據遷移也可以使用這種方法。首先將HDFS中的數據文件通過MapReduce任務生成HFile文件，然后將HFile文件導入HBase數據表（該數據表已存在）。


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 HBase rowkey設計實例 java操作Hbase實例 HBase的rowkey設計（含實例） HBase的rowkey設計（含實例） 2、通過HBase API進行開發 HBase(一): c#訪問hbase組件開發 Hbase之使用多Get實例返回數據 使用eclipse開發hbase程序 HBase開發篇 _ 一文說清HBase Connection的使用 調試intellij IDEA hbase開發環境
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM