浅谈HBase在SpringBoot项目里的应用(含HBaseUtil工具类)

浅谈HBase在SpringBoot项目里的应用(含HBaseUtil工具类)

背景：

项目这两个月开始使用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-client

1.3.0

org.slf4j

slf4j-log4j12

log4j

log4j

javax.servlet

servlet-api

org.apache.hadoop

hadoop-common

2.6.0

org.apache.hadoop

hadoop-mapreduce-client-core

2.6.0

org.apache.hadoop

hadoop-mapreduce-client-common

2.6.0

org.apache.hadoop

hadoop-hdfs

2.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;

/**

* Hbase-Conf配置

*

* @Author: yuanj

* @Date: 2018/10/12 10:49

*/

@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;

/**

* Spring的ApplicationContext的持有者,可以用静态方法的方式获取spring容器中的bean

*/

@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)http:// {

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;

}

/**

* 创建表

*

* @param tableName 表名

* @param columnFamily 列族（数组）

*/

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);

}

}

/**

* 插入记录（单行单列族-多列多值）

*

* @param tableName 表名

* @param row 行名

* @param columnFamilys 列族名

* @param columns 列名（数组）

* @param values 值（数组）（且需要和列一一对应）

*/

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);

}

}

/**

* 插入记录（单行单列族-单列单值）

*

* @param tableName 表名

* @param row 行名

* @param columnFamily 列族名

* @param column 列名

* @param value 值

*/

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);

}

/**

* 删除一行记录

*

* @param tablename 表名

* @param rowkey 行名

*/

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);

}

/**

* 删除单行单列族记录

* @param tablename 表名

* @param rowkey 行名

* @param columnFamily 列族名

*/

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);

}

/**

* 删除单行单列族单列记录

*

* @param tablename 表名

* @param rowkey 行名

* @param columnFamily 列族名

* @param column 列名

*/

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);

}

/**

* 查找一行记录

*

* @param tablename 表名

* @param rowKey 行名

*/

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;

}

/**

* 查找单行单列族单列记录

*

* @param tablename 表名

* @param rowKey 行名

* @param columnFamily 列族名

* @param column 列名

* @return

*/

public static String selectValue(String tablename, String rowKey, Stringhttp:// 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());

}

/**

* 查询表中所有行（Scan方式）

*

* @param tablename

* @return

*/

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;

}

/**

* 根据rowkey关键字查询报告记录

*

* @param tablename

* @param rowKeyword

* @return

*/

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;

}

/**

* 根据rowkey关键字和时间戳范围查询报告记录

*

* @param tablename

* @param rowKeyword

* @return

*/

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;

}

/**

* 删除表操作

*

* @param tablename

*/

public void deleteTable(String tablename) throws IOException {

TableName name=TableName.valueOf(tablename);

if(admin.tableExists(name)) {

admin.disableTable(name);

admin.deleteTable(name);

}

}

/**

* 利用协处理器进行全表count统计

*

* @param tablename

*/

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;

/**

* @Author wangqiubao

* @Date 2019/9/24 15:28

* @Description

**/

@Configuration

public class UcareHbaseConfiguration {

/**

* 读取HBase的zookeeper地址

*/

@Value("${hbase.zookeeper.quorum}")

private String quorum;

/**

* 配置HBase连接参数

*

* @return

*/

@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));

}

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。