背景：
项目这两个月开始使用Hbase来读写数据，网上现成的Hbase工具类要么版本混杂，要么只是Demo级别的简单实现，各方面都不完善；
而且我发现Hbase查询有很多种方式，首先大方向上有 Get 和 Scan两种，其次行键、列族、列名(限定符)、列值(value)、时间戳版本等多种组合条件，还有各种过滤器的选择，协处理器的应用，所以必须根据自己项目需求和Hbase行列设计来自定义Hbase工具类和实现类！
经过我自己的研究整理，在此分享下初步的实现方案吧 ~
注：Hbase版本：1.3.0 - CDH5.13.0 、SpringBoot版本：1.5.9
需要注意的是我用的是原生api，没有用和spring或者springboot整合的HbaseTemplate等，因为这方面资料较少而且听说并没有那么好用…
一、pom.xml 依赖
org.apache.hbase hbase-client1.3.0 org.slf4j slf4j-log4j12log4j log4jjavax.servlet servlet-apiorg.apache.hadoop hadoop-common2.6.0 org.apache.hadoop hadoop-mapreduce-client-core2.6.0 org.apache.hadoop hadoop-mapreduce-client-common2.6.0 org.apache.hadoop hadoop-hdfs2.6.0
二、application.yml 项目配置
此处我是自定义Hbase配置，后面会有专门的配置类来加载这个配置
hbase:
conf:
confMaps:
'hbase.zookeeper.quorum' : 'cdh1:2181,cdh2:2181,cdh3:2181'
三、HbaseConfig 自定义配置类
HbaseConfig.java：
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import java.util.Map;
@Configuration
@ConfigurationProperties(prefix = HbaseConfig.CONF_PREFIX)
public class HbaseConfig {
public static final String CONF_PREFIX = "hbase.conf";
private Map confMaps;
public Map getconfMaps() {
return confMaps;
}
public void setconfMaps(Map confMaps) {
this.confMaps = confMaps;
}
}
不了解@ConfigurationProperties这个注解的兄弟可以去百度下，它可以将application.yml中的配置导入到该类的成员变量里！
也就是说springboot项目启动完成后 confMaps变量里已经存在一个key为 hbase.zookeeper.quorum ，value为 cdh1:2181,cdh2:2181,cdh3:2181的entry了！
四、HbaseUtils工具类
首先添加 SpringContextHolder 工具类，下面会用到：
package com.moerlong.credit.core;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
@Component
public class SpringContextHolder implements ApplicationContextAware {
private static ApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
SpringContextHolder.applicationContext = applicationContext;
}
public static ApplicationContext getApplicationContext() {
assertApplicationContext();
return applicationContext;
}
@SuppressWarnings("unchecked")
public static T getBean(String beanName) {
assertApplicationContext();
return (T) applicationContext.getBean(beanName);
}
public static T getBean(Class requiredType) {
assertApplicationContext();
return applicationContext.getBean(requiredType);
}
private static void assertApplicationContext() {
if (SpringContextHolder.applicationContext == null) {
throw new RuntimeException("applicaitonContext属性为null,请检查是否注入了SpringContextHolder!");
}
}
}
HbaseUtils .java：
import com.moerlong.credit.config.HbaseConfig;
import com.moerlong.credit.core.SpringContextHolder;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@DependsOn("springContextHolder") //控制依赖顺序，保证springContextHolder类在之前已经加载
@Component
public class HbaseUtils {
private Logger logger = LoggerFactory.getLogger(this.getClass());
//手动获取hbaseConfig配置类对象
private static HbaseConfig hbaseConfig = SpringContextHolder.getBean("hbaseConfig");
private static Configuration conf = HbaseConfiguration.create();
private static ExecutorService pool = Executors.newScheduledThreadPool(20); //设置连接池
private static Connection connection = null;
private static HbaseUtils instance = null;
private static Admin admin = null;
private HbaseUtils(){
if(connection == null){
try {
//将hbase配置类中定义的配置加载到连接池中每个连接里
Map confMap = hbaseConfig.getconfMaps();
for (Map.Entry confEntry : confMap.entrySet()) {
conf.set(confEntry.getKey(), confEntry.getValue());
}
connection = ConnectionFactory.createConnection(conf, pool);
admin = connection.getAdmin();
} catch (IOException e) {
logger.error("HbaseUtils实例初始化失败！错误信息为：" + e.getMessage(), e);
}
}
}
//简单单例方法，如果autowired自动注入就不需要此方法
public static synchronized HbaseUtils getInstance(){
if(instance == null){
instance = new HbaseUtils();
}
return instance;
}
public void createTable(String tableName, String[] columnFamily) throws IOException{
TableName name = TableName.valueOf(tableName);
//如果存在则删除
if (admin.tableExists(name)) {
admin.disableTable(name);
admin.deleteTable(name);
logger.error("create htable error! this table {} already exists!", name);
} else {
HTableDescriptor desc = new HTableDescriptor(name);
for (String cf : columnFamily) {
desc.addFamily(new HColumnDescriptor(cf));
}
admin.createTable(desc);
}
}
public void insertRecords(String tableName, String row, String columnFamilys, String[] columns, String[] values) throws IOException {
TableName name = TableName.valueOf(tableName);
Table table = connection.getTable(name);
Put put = new Put(Bytes.toBytes(row));
for (int i = 0; i < columns.length; i++) {
put.addColumn(Bytes.toBytes(columnFamilys), Bytes.toBytes(columns[i]), Bytes.toBytes(values[i]));
table.put(put);
}
}
public void insertOneRecord(String tableName, String row, String columnFamily, String column, String value) throws IOException {
TableName name = TableName.valueOf(tableName);
Table table = connection.getTable(name);
Put put = new Put(Bytes.toBytes(row));
put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
table.put(put);
}
public void deleteRow(String tablename, String rowkey) throws IOException {
TableName name = TableName.valueOf(tablename);
Table table = connection.getTable(name);
Delete d = new Delete(rowkey.getBytes());
table.delete(d);
}
public void deleteColumnFamily(String tablename, String rowkey, String columnFamily) throws IOException {
TableName name = TableName.valueOf(tablename);
Table table = connection.getTable(name);
Delete d = new Delete(rowkey.getBytes()).deleteFamily(Bytes.toBytes(columnFamily));
table.delete(d);
}
public void deleteColumn(String tablename, String rowkey, String columnFamily, String column) throws IOException {
TableName name = TableName.valueOf(tablename);
Table table = connection.getTable(name);
Delete d = new Delete(rowkey.getBytes()).deleteColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
table.delete(d);
}
public static String selectRow(String tablename, String rowKey) throws IOException {
String record = "";
TableName name=TableName.valueOf(tablename);
Table table = connection.getTable(name);
Get g = new Get(rowKey.getBytes());
Result rs = table.get(g);
NavigableMap>> map = rs.getMap();
for (Cell cell : rs.rawCells()) {
StringBuffer stringBuffer = new StringBuffer().append(Bytes.toString(cell.getRow())).append("t")
.append(Bytes.toString(cell.getFamily())).append("t")
.append(Bytes.toString(cell.getQualifier())).append("t")
.append(Bytes.toString(cell.getValue())).append("n");
String str = stringBuffer.toString();
record += str;
}
return record;
}
public static String selectValue(String tablename, String rowKey, String columnFamily, String column) throws IOException {
TableName name=TableName.valueOf(tablename);
Table table = connection.getTable(name);
Get g = new Get(rowKey.getBytes());
g.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
Result rs = table.get(g);
return Bytes.toString(rs.value());
}
public String scanAllRecord(String tablename) throws IOException {
String record = "";
TableName name=TableName.valueOf(tablename);
Table table = connection.getTable(name);
Scan scan = new Scan();
ResultScanner scanner = table.getScanner(scan);
try {
for(Result result : scanner){
for (Cell cell : result.rawCells()) {
StringBuffer stringBuffer = new StringBuffer().append(Bytes.toString(cell.getRow())).append("t")
.append(Bytes.toString(cell.getFamily())).append("t")
.append(Bytes.toString(cell.getQualifier())).append("t")
.append(Bytes.toString(cell.getValue())).append("n");
String str = stringBuffer.toString();
record += str;
}
}
} finally {
if (scanner != null) {
scanner.close();
}
}
return record;
}
public List scanReportDataByRowKeyword(String tablename, String rowKeyword) throws IOException {
ArrayList<> list = new ArrayList<>();
Table table = connection.getTable(TableName.valueOf(tablename));
Scan scan = new Scan();
//添加行键过滤器，根据关键字匹配
RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(rowKeyword));
scan.setFilter(rowFilter);
ResultScanner scanner = table.getScanner(scan);
try {
for (Result result : scanner) {
//TODO 此处根据业务来自定义实现
list.add(null);
}
} finally {
if (scanner != null) {
scanner.close();
}
}
return list;
}
public List scanReportDataByRowKeywordTimestamp(String tablename, String rowKeyword, Long minStamp, Long maxStamp) throws IOException {
ArrayList<> list = new ArrayList<>();
Table table = connection.getTable(TableName.valueOf(tablename));
Scan scan = new Scan();
//添加scan的时间范围
scan.setTimeRange(minStamp, maxStamp);
RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(rowKeyword));
scan.setFilter(rowFilter);
ResultScanner scanner = table.getScanner(scan);
try {
for (Result result : scanner) {
//TODO 此处根据业务来自定义实现
list.add(null);
}
} finally {
if (scanner != null) {
scanner.close();
}
}
return list;
}
public void deleteTable(String tablename) throws IOException {
TableName name=TableName.valueOf(tablename);
if(admin.tableExists(name)) {
admin.disableTable(name);
admin.deleteTable(name);
}
}
public Long countRowsWithCoprocessor(String tablename) throws Throwable {
TableName name=TableName.valueOf(tablename);
HTableDescriptor descriptor = admin.getTableDescriptor(name);
String coprocessorClass = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";
if (! descriptor.hasCoprocessor(coprocessorClass)) {
admin.disableTable(name);
descriptor.addCoprocessor(coprocessorClass);
admin.modifyTable(name, descriptor);
admin.enableTable(name);
}
//计时
StopWatch stopWatch = new StopWatch();
stopWatch.start();
Scan scan = new Scan();
AggregationClient aggregationClient = new AggregationClient(conf);
Long count = aggregationClient.rowCount(name, new LongColumnInterpreter(), scan);
stopWatch.stop();
System.out.println("RowCount：" + count + "，全表count统计耗时：" + stopWatch.getTotalTimeMillis());
return count;
}
}
五、使用
接下来只需要在项目业务类里注入hbaseUtils就可以使用了：
@Autowired
private HbaseUtils hbaseUtils;
补充知识：springboot整合Hbase
springboot项目需要整合SpringCloud
依赖
org.apache.hbase
hbase-shaded-client
1.2.6
yml配置:
自定义配置读取zookeeper配置
hbase:
zookeeper:
quorum: hbase126-node2:2181
config配置:
import net.cc.commons.exception.CCRuntimeException;
import org.apache.hadoop.hbase.HbaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import java.io.IOException;
import java.util.function.Supplier;
@Configuration
public class UcareHbaseConfiguration {
@Value("${hbase.zookeeper.quorum}")
private String quorum;
@Bean
public org.apache.hadoop.conf.Configuration hbaseConfig() {
org.apache.hadoop.conf.Configuration config = HbaseConfiguration.create();
config.set(HConstants.ZOOKEEPER_QUORUM, quorum);
return config;
}
//每次调用get方法就会创建一个Connection
@Bean
public Supplier hbaseConnSupplier() {
return () -> {
try {
return hbaseConnection();
} catch (IOException e) {
throw new CCRuntimeException(e);
}
};
}
@Bean
//@Scope标明模式,默认单例模式. prototype多例模式
//若是在其他类中直接@Autowired引入的,多例就无效了,因为那个类在初始化的时候,已经创建了创建了这个bean了,之后调用的时候,不会重新创建,若是想要实现多例,就要每次调用的时候,手动获取bean
@Scope(value = "prototype")
public Connection hbaseConnection() throws IOException {
return ConnectionFactory.createConnection(hbaseConfig());
}
}
使用
spring管理
@Autowired private Connection hbaseConnection;
插入/更新数据
public void aaaa() throws IOException {
try (Table table = hbaseConnection.getTable(TableName.valueOf("表名"))) {//获取表连接
//配置一条数据
// 行键
Put put = new Put(Bytes.toBytes("key主键"));
put.addColumn(Bytes.toBytes("列族"), Bytes.toBytes("列"), Bytes.toBytes("值"));
.....//每个有数据的列都要一个addColumn
//put插入数据
table.put(put);
}
}
查询
根据主键查询内容
try (Table table = hbaseConnection.getTable(TableName.valueOf("表名"))) {
Result result = table.get(new Get(asRowKey(date, acid)));
if (result == null) return null;
// 列名为starttime，最后一条就是该航班最新的航迹
Cell latestCell = Iterables.getLast(result.listCells());
return AdsbTrackProto.AdsbTrack.parseFrom(CellUtil.clonevalue(latestCell));
}
以上这篇浅谈Hbase在SpringBoot项目里的应用(含HbaseUtil工具类)就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持考高分网。
