Hbase的基本结构的代码
package hbase1;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
public class HbaseConnector {
private Integer tablePoolSize;
private HTablePool tablePool;
public void init(){
Configuration configuration= HBaseConfiguration.create();
tablePool=new HTablePool(configuration,tablePoolSize);
}
public HTableInterface getTable(String tableName){
return tablePool.getTable(tableName);
}
}
package hbase1;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.springframework.util.StringUtils;
import java.io.IOException;
public class HbaseExecutorTemplate {
public <T> T execute(HbaseConnector hBaseConnector, String hbaseTableName, HbaseOpMode hbaseOpMode, HBaseOperationCallBack<T> action) {
HTableInterface hbaseTable = null;
boolean success = true;
long start = System.currentTimeMillis();
try {
if (hBaseConnector == null) {
throw new Exception("can not find hbaseConnectior");
}
if (StringUtils.isEmpty(hbaseTableName)) {
throw new Exception("chbaseTableName is blank");
}
hbaseTable = hBaseConnector.getTable(hbaseTableName);
if (hbaseTable == null) {
throw new Exception("hbaseTable is blank");
}
T result = action.hCall(hbaseTable);
return result;
} catch (Exception e) {
success = false;
}finally {
try{
if(hbaseTable!=null){
hbaseTable.close();
}
}catch (Exception e){
}
}
return null;
}
}
package hbase1;
import java.util.List;
public interface HbaseManager {
HbaseRow queryLatestRow(String tableName, String rowKey, List<String> qualifiers);
}
package hbase1;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
public class HbaseManagerImpl implements HbaseManager {
/**
* 建议 habse 一个表 一个列族,多个qualifier
*/
private String family="family";
private HbaseConnector hbaseConnector;
private HbaseExecutorTemplate hbaseExecutorTemplate;
@Override
public HbaseRow queryLatestRow(String tableName, String rowKey, List<String> qualifiers) {
return hbaseExecutorTemplate.execute(hbaseConnector, tableName, HbaseOpMode.get, new HBaseOperationCallBack<HbaseRow>() {
@Override
public HbaseRow hCall(HTableInterface hTable) throws Exception {
Get hGet=new Get(Bytes.toBytes(rowKey));
hGet.setMaxVersions();
qualifiers.forEach(qualifier->hGet.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier)));
Result result=hTable.get(hGet);
if(result==null){
return null;
}
Map<String,String> qualifierValMap=new HashMap<>();
Map<String,Long> qualifierTimestampMap=new HashMap<>();
result.list().stream().filter(Objects::nonNull).forEach(kv->{
String qualifier=Bytes.toString(kv.getQualifier());
String value=Bytes.toString(kv.getValue());
Long timeStamp=kv.getTimestamp();
qualifierValMap.put(qualifier,value);
qualifierTimestampMap.put(qualifier,timeStamp);
});
HbaseRow hbaseRow=new HbaseRow();
hbaseRow.setTable(tableName);
hbaseRow.setFamily(family);
hbaseRow.setRowKey(rowKey);
hbaseRow.setQualifierTimestampMap(qualifierTimestampMap);
hbaseRow.setQualifierValMap(qualifierValMap);
return hbaseRow;
}
});
}
}
package hbase1;
import org.apache.hadoop.hbase.client.HTableInterface;
public interface HBaseOperationCallBack<T> {
T hCall(HTableInterface hTableInterface) throws Exception;
}
package hbase1;
public enum HbaseOpMode {
getTable("getTable"),
put("put"),
puts("puts"),
get("get"),
gets("gets"),
delete("delete"),
deletes("deletes"),
scan("scan"),
checkAndPut("checkAndPut"),
incrementColumnValue("incrementColumnValue")
;
private String name;
HbaseOpMode(String name) {
this.name = name;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
package hbase1;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
public class HbaseRow implements Serializable {
private String table;
private String family;
private String rowKey;
/**
* 不带多版本列名
* qualifier-》value
*/
private Map<String,String> qualifierValMap=new HashMap<>();
/**
* 带多版本qualifier-》(version-》value)
*/
private Map<String,Map<String,String>> qualifierVersionValMap=new HashMap<>();
/**
* 时间戳 qualifier-》timestamp
*/
private Map<String,Long> qualifierTimestampMap=new HashMap<>();
public String getTable() {
return table;
}
public void setTable(String table) {
this.table = table;
}
public String getFamily() {
return family;
}
public void setFamily(String family) {
this.family = family;
}
public String getRowKey() {
return rowKey;
}
public void setRowKey(String rowKey) {
this.rowKey = rowKey;
}
public Map<String, String> getQualifierValMap() {
return qualifierValMap;
}
public void setQualifierValMap(Map<String, String> qualifierValMap) {
this.qualifierValMap = qualifierValMap;
}
public Map<String, Map<String, String>> getQualifierVersionValMap() {
return qualifierVersionValMap;
}
public void setQualifierVersionValMap(Map<String, Map<String, String>> qualifierVersionValMap) {
this.qualifierVersionValMap = qualifierVersionValMap;
}
public Map<String, Long> getQualifierTimestampMap() {
return qualifierTimestampMap;
}
public void setQualifierTimestampMap(Map<String, Long> qualifierTimestampMap) {
this.qualifierTimestampMap = qualifierTimestampMap;
}
}
package hbase1;
public interface HistoryRecordService {
String queryHistoryRecordByUid(String uid);
}
package hbase1;
import org.springframework.util.DigestUtils;
import java.util.ArrayList;
import java.util.List;
public class HistoryRecordServiceImpl implements HistoryRecordService {
private String tableName;
private HbaseManager hbaseManager;
@Override
public String queryHistoryRecordByUid(String uid) {
final String QUALIFIER="record_id_list";
HbaseRow hbaseRow=null;
try{
List<String> quailfiers=new ArrayList<>();
quailfiers.add(QUALIFIER);
String rowKey= DigestUtils.md5DigestAsHex(uid.getBytes()).substring(0,4)+uid;
hbaseRow=hbaseManager.queryLatestRow(tableName,rowKey,quailfiers);
}catch (Exception e){
}
if(hbaseRow==null){
return null;
}
if(hbaseRow.getQualifierValMap()!=null){
return hbaseRow.getQualifierValMap().get(QUALIFIER);
}
return null;
}
}