淺談HBase在SpringBoot項(xiàng)目里的應(yīng)用(含HBaseUtil工具類(lèi))
更新時(shí)間：2020年10月21日 10:02:39 作者：深寒丶
這篇文章主要介紹了淺談HBase在SpringBoot項(xiàng)目里的應(yīng)用(含HBaseUtil工具類(lèi))，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧
背景：
項(xiàng)目這兩個(gè)月開(kāi)始使用HBase來(lái)讀寫(xiě)數(shù)據(jù)，網(wǎng)上現(xiàn)成的HBase工具類(lèi)要么版本混雜，要么只是Demo級(jí)別的簡(jiǎn)單實(shí)現(xiàn)，各方面都不完善；
而且我發(fā)現(xiàn)HBase查詢(xún)有很多種方式，首先大方向上有 Get 和 Scan兩種，其次行鍵、列族、列名(限定符)、列值(value)、時(shí)間戳版本等多種組合條件，還有各種過(guò)濾器的選擇，協(xié)處理器的應(yīng)用，所以必須根據(jù)自己項(xiàng)目需求和HBase行列設(shè)計(jì)來(lái)自定義HBase工具類(lèi)和實(shí)現(xiàn)類(lèi)！
經(jīng)過(guò)我自己的研究整理，在此分享下初步的實(shí)現(xiàn)方案吧 ~
注：HBase版本：1.3.0 - CDH5.13.0 、SpringBoot版本：1.5.9
需要注意的是我用的是原生api，沒(méi)有用和spring或者springboot整合的HbaseTemplate等，因?yàn)檫@方面資料較少而且聽(tīng)說(shuō)并沒(méi)有那么好用…
一、pom.xml 依賴(lài)
<dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>1.3.0</version> <exclusions> <exclusion> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> <exclusion> <groupId>log4j</groupId> <artifactId>log4j</artifactId> </exclusion> <exclusion> <groupId>javax.servlet</groupId> <artifactId>servlet-api</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.6.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-core</artifactId> <version>2.6.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-common</artifactId> <version>2.6.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.6.0</version> </dependency>
二、application.yml 項(xiàng)目配置
此處我是自定義HBase配置，后面會(huì)有專(zhuān)門(mén)的配置類(lèi)來(lái)加載這個(gè)配置
hbase:
conf:
confMaps:
'hbase.zookeeper.quorum' : 'cdh1:2181,cdh2:2181,cdh3:2181'
三、HbaseConfig 自定義配置類(lèi)
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<String,String> confMaps;
public Map<String, String> getconfMaps() {
return confMaps;
}
public void setconfMaps(Map<String, String> confMaps) {
this.confMaps = confMaps;
}
}
不了解@ConfigurationProperties這個(gè)注解的兄弟可以去百度下，它可以將application.yml中的配置導(dǎo)入到該類(lèi)的成員變量里！
也就是說(shuō)springboot項(xiàng)目啟動(dòng)完成后 confMaps變量里已經(jīng)存在一個(gè)key為 hbase.zookeeper.quorum ，value為 cdh1:2181,cdh2:2181,cdh3:2181的entry了！
四、HBaseUtils工具類(lèi)
首先添加 SpringContextHolder 工具類(lèi)，下面會(huì)用到：
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的持有者,可以用靜態(tài)方法的方式獲取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> T getBean(String beanName) {
assertApplicationContext();
return (T) applicationContext.getBean(beanName);
}
public static <T> T getBean(Class<T> requiredType) {
assertApplicationContext();
return applicationContext.getBean(requiredType);
}
private static void assertApplicationContext() {
if (SpringContextHolder.applicationContext == null) {
throw new RuntimeException("applicaitonContext屬性為null,請(qǐng)檢查是否注入了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") //控制依賴(lài)順序，保證springContextHolder類(lèi)在之前已經(jīng)加載
@Component
public class HBaseUtils {
private Logger logger = LoggerFactory.getLogger(this.getClass());
//手動(dòng)獲取hbaseConfig配置類(lèi)對(duì)象
private static HbaseConfig hbaseConfig = SpringContextHolder.getBean("hbaseConfig");
private static Configuration conf = HBaseConfiguration.create();
private static ExecutorService pool = Executors.newScheduledThreadPool(20); //設(shè)置連接池
private static Connection connection = null;
private static HBaseUtils instance = null;
private static Admin admin = null;
private HBaseUtils(){
if(connection == null){
try {
//將hbase配置類(lèi)中定義的配置加載到連接池中每個(gè)連接里
Map<String, String> confMap = hbaseConfig.getconfMaps();
for (Map.Entry<String,String> confEntry : confMap.entrySet()) {
conf.set(confEntry.getKey(), confEntry.getValue());
}
connection = ConnectionFactory.createConnection(conf, pool);
admin = connection.getAdmin();
} catch (IOException e) {
logger.error("HbaseUtils實(shí)例初始化失?。″e(cuò)誤信息為：" + e.getMessage(), e);
}
}
}
//簡(jiǎn)單單例方法，如果autowired自動(dòng)注入就不需要此方法
public static synchronized HBaseUtils getInstance(){
if(instance == null){
instance = new HBaseUtils();
}
return instance;
}
/**
* 創(chuàng)建表
*
* @param tableName 表名
* @param columnFamily 列族（數(shù)組）
*/
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 列名（數(shù)組）
* @param values 值（數(shù)組）（且需要和列一一對(duì)應(yīng)）
*/
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<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> 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, 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());
}
/**
* 查詢(xún)表中所有行（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;
}
/**
* 根據(jù)rowkey關(guān)鍵字查詢(xún)報(bào)告記錄
*
* @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();
//添加行鍵過(guò)濾器，根據(jù)關(guān)鍵字匹配
RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(rowKeyword));
scan.setFilter(rowFilter);
ResultScanner scanner = table.getScanner(scan);
try {
for (Result result : scanner) {
//TODO 此處根據(jù)業(yè)務(wù)來(lái)自定義實(shí)現(xiàn)
list.add(null);
}
} finally {
if (scanner != null) {
scanner.close();
}
}
return list;
}
/**
* 根據(jù)rowkey關(guān)鍵字和時(shí)間戳范圍查詢(xún)報(bào)告記錄
*
* @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的時(shí)間范圍
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 此處根據(jù)業(yè)務(wù)來(lái)自定義實(shí)現(xiàn)
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);
}
}
/**
* 利用協(xié)處理器進(jìn)行全表count統(tǒng)計(jì)
*
* @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);
}
//計(jì)時(shí)
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統(tǒng)計(jì)耗時(shí)：" + stopWatch.getTotalTimeMillis());
return count;
}
}
五、使用
接下來(lái)只需要在項(xiàng)目業(yè)務(wù)類(lèi)里注入hbaseUtils就可以使用了：
@Autowired
private HBaseUtils hBaseUtils;
補(bǔ)充知識(shí)：springboot整合Hbase
springboot項(xiàng)目需要整合SpringCloud
依賴(lài)
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-shaded-client</artifactId>
<version>1.2.6</version>
</dependency>
<!---->
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連接參數(shù)
*
* @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;
}
//每次調(diào)用get方法就會(huì)創(chuàng)建一個(gè)Connection
@Bean
public Supplier<Connection> hbaseConnSupplier() {
return () -> {
try {
return hbaseConnection();
} catch (IOException e) {
throw new CCRuntimeException(e);
}
};
}
@Bean
//@Scope標(biāo)明模式,默認(rèn)單例模式. prototype多例模式
//若是在其他類(lèi)中直接@Autowired引入的,多例就無(wú)效了,因?yàn)槟莻€(gè)類(lèi)在初始化的時(shí)候,已經(jīng)創(chuàng)建了創(chuàng)建了這個(gè)bean了,之后調(diào)用的時(shí)候,不會(huì)重新創(chuàng)建,若是想要實(shí)現(xiàn)多例,就要每次調(diào)用的時(shí)候,手動(dòng)獲取bean
@Scope(value = "prototype")
public Connection hbaseConnection() throws IOException {
return ConnectionFactory.createConnection(hbaseConfig());
}
}
使用
spring管理
/** * 內(nèi)部已實(shí)現(xiàn)線(xiàn)程安全的連接池 */ @Autowired private Connection hbaseConnection;
插入/更新數(shù)據(jù)
public void aaaa() throws IOException {
try (Table table = hbaseConnection.getTable(TableName.valueOf("表名"))) {//獲取表連接
//配置一條數(shù)據(jù)
// 行鍵
Put put = new Put(Bytes.toBytes("key主鍵"));
put.addColumn(Bytes.toBytes("列族"), Bytes.toBytes("列"), Bytes.toBytes("值"));
.....//每個(gè)有數(shù)據(jù)的列都要一個(gè)addColumn
//put插入數(shù)據(jù)
table.put(put);
}
}
查詢(xún)
根據(jù)主鍵查詢(xún)內(nèi)容
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項(xiàng)目里的應(yīng)用(含HBaseUtil工具類(lèi))就是小編分享給大家的全部?jī)?nèi)容了，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。
相關(guān)文章
Java @PostMapping和@GetMapping方法使用詳解
這篇文章主要介紹了Java @PostMapping和@GetMapping方法使用，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)吧2023-03-03
java Hibernate save()與persist()區(qū)別
本文章來(lái)給各位同學(xué)介紹一下Hibernate save()與persist()區(qū)別,希望此文章能對(duì)各位同學(xué)對(duì)于Hibernate save()與persist()有所理解2016-01-01
Spring Data JPA 之 JpaRepository的使用
這篇文章主要介紹了Spring Data JPA 之 JpaRepository的使用方式，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-02-02
SpringBoot Starter依賴(lài)原理與實(shí)例詳解
SpringBoot中的starter是一種非常重要的機(jī)制，能夠拋棄以前繁雜的配置，將其統(tǒng)一集成進(jìn)starter，應(yīng)用者只需要在maven中引入starter依賴(lài)，SpringBoot就能自動(dòng)掃描到要加載的信息并啟動(dòng)相應(yīng)的默認(rèn)配置。starter讓我們擺脫了各種依賴(lài)庫(kù)的處理，需要配置各種信息的困擾2022-09-09
這篇文章主要介紹了Spring和Mybatis整合的原理詳解，文章圍繞主題展開(kāi)詳細(xì)的內(nèi)容介紹，具有一定的參考價(jià)值，需要的小伙伴可以參考一下2022-07-07
這篇文章主要給大家介紹了關(guān)于java整合onlyoffice的各種踩坑,OnlyOffice是一種強(qiáng)大的在線(xiàn)協(xié)作軟件,專(zhuān)為企業(yè)和個(gè)人設(shè)計(jì),文中通過(guò)代碼介紹的非常詳細(xì),需要的朋友可以參考下2023-07-07
SpringCloud Config連接git與數(shù)據(jù)庫(kù)流程分析講解
springcloud config是一個(gè)解決分布式系統(tǒng)的配置管理方案。它包含了 client和server兩個(gè)部分，server端提供配置文件的存儲(chǔ)、以接口的形式將配置文件的內(nèi)容提供出去，client端通過(guò)接口獲取數(shù)據(jù)、并依據(jù)此數(shù)據(jù)初始化自己的應(yīng)用2022-12-12
Java 如何使用Feign發(fā)送HTTP請(qǐng)求
這篇文章主要介紹了Java 如何使用Feign發(fā)送HTTP請(qǐng)求，幫助大家更好的理解和學(xué)習(xí)Java，感興趣的朋友可以了解下2020-11-11
這篇文章主要為大家詳細(xì)介紹了java中Properties文件加載和使用方法，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-03-03