HBase2 java Api介面說明和程式碼實、支援kerberos認證
阿新 • • 發佈：2021-02-06
文章目錄
說明
本博文主要介紹hbase2 （版本1和2介面略有不同）版本java API相關操作，主要涉及介面介紹和名稱空間、表、資料操作。
hbase jar包使用log4j日誌系統，未找到切換方式，建議不要改用logback。
介面介紹
攔截器
- CompareOperator.LESS 匹配小宇設定值的值
- CompareOperator.LESS_OR_EQUAL 匹配小宇或等於設定值的值
- CompareOperator.EQUAL 匹配等於設定值的值
- CompareOperator.NOT_EQUAL 匹配與設定值不相等的值
- CompareOperator.GREATER_OR_EQUAL 匹配大於或等於設定值的值
- CompareOperator.GREATER 匹配大於設定值的值 CompareOperator.NO_OP 排除一切值
比較器種類
- RowFilter ：基於行鍵來過濾資料；
- FamilyFilterr ：基於列族來過濾資料；
- QualifierFilterr ：基於列限定符（列名）來過濾資料；
- ValueFilterr ：基於單元格 (cell) 的值來過濾資料；
- DependentColumnFilter ：指定一個參考列來過濾其他列的過濾器，過濾的原則是基於參考列的時間戳來進行篩選 。
比較器
- BinaryComparator 使用Bytes.compareTo()比較當前值與閾值
- BinaryPrefixComparator 與上面類似，但是是從左端開始字首匹配
- NullComparator 不做匹配，只判斷當前值是不是null
- BitComparator 通過BitwiseOp類提供的按位與（AND）、或（OR）、異或（XOR）操作執行位級比較 ，只能用EQUAL和NOT_EQUAL
- RegexStringComparator 根據一個正則表示式，在例項化這個比較器的時候去匹配表中的資料 ，只能用* EQUAL和NOT_EQUAL
- SubStringComparator 把閾值和表中資料當做String例項，同時通過contains()操作匹配字串，只能用EQUAL和NOT_EQUAL
專用過濾器
- 單列列值過濾器 (SingleColumnValueFilter)
- 單列列值排除器 (SingleColumnValueExcludeFilter)
- 行鍵字首過濾器 (PrefixFilter)
- 列名字首過濾器 (ColumnPrefixFilter)
- 分頁過濾器 (PageFilter)
- 時間戳過濾器 (TimestampsFilter)
- 首次行鍵過濾器 (FirstKeyOnlyFilter)
包裝過濾器
- SkipFilter過濾器，遇到需要過濾keyvalue例項時，拓張過濾整行資料
- WhileMatchFilter過濾器 遇到一個需要過濾的 KeyValue 例項時，WhileMatchFilter 則結束本次掃描，返回已經掃描到的結果
- FilterList 過濾器類組合，多種型別過濾器組合。
java程式碼
maven導包
- 日誌介面框架使用slf4j，這裡去除commons-logging。
<!-- hbase客戶端 -->
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>2.0.2</version>
<!--剔除commons-logging-->
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
java程式碼
- 實現對hbase庫資料的增刪改查，支援kerberos認證，為避免複雜引數設定，這裡直接引入hadoop和hbase配置檔案。
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public final class HbaseUtil {
private static Logger log= LoggerFactory.getLogger(HbaseUtil.class);
private Connection connection;
public HbaseUtil() throws IOException {
Configuration conf = HBaseConfiguration.create();
conf.addResource(new Path(ConfigUtil.hbaseFile));
conf.addResource(new Path(ConfigUtil.coreFile));
conf.addResource(new Path(ConfigUtil.hdfsFile));
if(ConfigUtil.kerberos==1) {
System.setProperty("java.security.krb5.conf", "/etc/krb5.conf");
UserGroupInformation.setConfiguration(conf);
UserGroupInformation.loginUserFromKeytab(ConfigUtil.kerberosUser,ConfigUtil.kerberosFile);
HBaseAdmin.available(conf);
}
this.connection=ConnectionFactory.createConnection(conf);
}
/**
* 範圍查詢
* @param tableName 表名
* @param startRowkey 開始rowkey
* @param endRowkey 結束rowkey不查詢
* @return
*/
public List<String> getData(String tableName,String startRowkey,String endRowkey){
log.info("Search Table {} ,Startrowkey:{} ,Endrowkey:{}",tableName,startRowkey,endRowkey);
List<String> dataList=new ArrayList<>();
ResultScanner resultList = null;
String rowkey;
String filterString=startRowkey.substring(4);
try {
Table tableModel=connection.getTable(TableName.valueOf(tableName));
Scan scan = new Scan();
//新增start和end
scan.withStartRow(Bytes.toBytes(startRowkey));
scan.withStopRow(Bytes.toBytes(endRowkey));
scan.addColumn(Bytes.toBytes(ConfigUtil.familyName), Bytes.toBytes(ConfigUtil.cloumnName));
resultList = tableModel.getScanner(scan);
if(resultList!=null) {
for (Result result : resultList) {
//TODO 新增rowkey規範驗證
rowkey = Bytes.toString(result.getValue(Bytes.toBytes(ConfigUtil.familyName), Bytes.toBytes(ConfigUtil.cloumnName)));
if(StringUtil.isNotEmpty(rowkey)) {
dataList.add(rowkey);
}
}
}
tableModel.close();
} catch (Exception e) {
log.error(e.toString(),e);
}
return dataList;
}
/**
* 根據rowkey批量查詢
* @param tableName 表名
* @param rowkeyList rowkey列表
* @return
*/
public List<String> getDataList(String tableName,List<String> rowkeyList){
log.info("Search Table {} ,rowkeyList:{} ",tableName,JsonUtil.toJson(rowkeyList));
List<String> dataList=new ArrayList<>();
try {
Table tableModel=connection.getTable(TableName.valueOf(tableName));
List<Get> getList=new ArrayList<>();
for(String rowkey:rowkeyList) {
getList.add(new Get(Bytes.toBytes(rowkey)));
}
//查詢
Result[] resultList=tableModel.get(getList);
//儲存資料
if(resultList!=null&&resultList.length>0) {
Cell[] cellList;
for(Result result:resultList) {
cellList=result.rawCells();
for(Cell cell:cellList) {
dataList.add(Bytes.toString(cell.getValueArray()));
}
}
}
tableModel.close();
} catch (Exception e) {
log.error(e.toString(),e);
}
return dataList;
}
/**
* 建立表
* @param tableName
*/
public boolean createTable(String tableName){
try {
//判斷資料庫是否存在
Admin admin=this.connection.getAdmin();
NamespaceDescriptor[] namespace=admin.listNamespaceDescriptors();
int state=0;
//獲取名稱空間
if(namespace.length>0) {
for(NamespaceDescriptor name:namespace){
if(name.getName().equals(ConfigUtil.dataName)){
state=1;
}
}
}
//建立名稱空間
if(state==0){
log.info("Create NameSpace {}",ConfigUtil.dataName);
admin.createNamespace(NamespaceDescriptor.create(ConfigUtil.dataName).build());
}
TableName table= TableName.valueOf(ConfigUtil.dataName+":"+tableName);
//建立表
if(admin.tableExists(table)){
log.info("{} tables Exists!",tableName);
}else{
log.info("Create Table {}",tableName);
//表描述器構造器
TableDescriptorBuilder tdb =TableDescriptorBuilder.newBuilder(table);
//列族描述起構造器
ColumnFamilyDescriptorBuilder cdb = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(ConfigUtil.familyName));
//儲存時間
cdb.setTimeToLive(ConfigUtil.saveTime*24*60*60);
//獲得列描述起
ColumnFamilyDescriptor cfd = cdb.build();
//新增列族
tdb.setColumnFamily(cfd);
//獲得表描述器
TableDescriptor td = tdb.build();
//建立表
admin.createTable(td);
log.info("{} Table Create Success!",tableName);
}
return true;
} catch (Exception e) {
log.error(e.toString(),e);
}
return false;
}
/**
* 判斷表是否存在
* @param tableName
* @return
*/
public boolean getTableStatus(String tableName){
try {
Admin admin=this.connection.getAdmin();
return admin.tableExists(TableName.valueOf(ConfigUtil.dataName+":"+tableName));
} catch (Exception e) {
log.error(e.toString(),e);
}
return false;
}
/**
* 刪除表
*/
public boolean delTable(String tableName){
try {
TableName table=TableName.valueOf(ConfigUtil.dataName+":"+tableName);
Admin admin=this.connection.getAdmin();
if(admin.tableExists(table)){
admin.disableTable(table);
admin.deleteTable(table);
log.info("Delete {} Success!",tableName);
}else{
log.info("No Found Table:{}",tableName);
}
return true;
} catch (Exception e) {
log.error(e.toString(),e);
}
return false;
}
/**
* 新增資料
* @param tableName
* @param data
*/
public void addData(String tableName,Map<String,String> data) {
try {
Table tableModel=connection.getTable(TableName.valueOf(ConfigUtil.dataName+":"+tableName));
List<Put> puts = new ArrayList<>();
Put put;
for(Map.Entry<String,String> entry:data.entrySet()) {
put= new Put(Bytes.toBytes(entry.getKey()));
put.addColumn(Bytes.toBytes(ConfigUtil.familyName),Bytes.toBytes(ConfigUtil.cloumnName), Bytes.toBytes(entry.getValue()));
puts.add(put);
}
tableModel.put(puts);
tableModel.close();
} catch (Exception e) {
log.error(e.toString(),e);
}
}
/**
* 關閉連線
*/
public void close() {
try {
this.connection.close();
} catch (IOException e) {
log.error(e.toString(),e);
}
}
}
總結
實際使用過程中發現，hbase API正則過濾器不能實現預設的規則過濾結果，java util引用正則過濾可以，hbase引擎失敗，建議scan後，進行二次程式碼邏輯處理，或直接程式碼邏輯處理。