淺談HBase在SpringBoot項(xiàng)目里的應(yīng)用(含HBaseUtil工具類)
背景:
項(xiàng)目這兩個(gè)月開(kāi)始使用HBase來(lái)讀寫(xiě)數(shù)據(jù),網(wǎng)上現(xiàn)成的HBase工具類要么版本混雜,要么只是Demo級(jí)別的簡(jiǎn)單實(shí)現(xiàn),各方面都不完善;
而且我發(fā)現(xiàn)HBase查詢有很多種方式,首先大方向上有 Get 和 Scan兩種,其次行鍵、列族、列名(限定符)、列值(value)、時(shí)間戳版本等多種組合條件,還有各種過(guò)濾器的選擇,協(xié)處理器的應(yīng)用,所以必須根據(jù)自己項(xiàng)目需求和HBase行列設(shè)計(jì)來(lái)自定義HBase工具類和實(shí)現(xiàn)類!
經(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 依賴
<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ì)有專門(mén)的配置類來(lái)加載這個(gè)配置
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<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)入到該類的成員變量里!
也就是說(shuō)springboot項(xiàng)目啟動(dòng)完成后 confMaps變量里已經(jīng)存在一個(gè)key為 hbase.zookeeper.quorum ,value為 cdh1:2181,cdh2:2181,cdh3:2181的entry了!
四、HBaseUtils工具類
首先添加 SpringContextHolder 工具類,下面會(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") //控制依賴順序,保證springContextHolder類在之前已經(jīng)加載 @Component public class HBaseUtils { private Logger logger = LoggerFactory.getLogger(this.getClass()); //手動(dòng)獲取hbaseConfig配置類對(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配置類中定義的配置加載到連接池中每個(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í)例初始化失敗!錯(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()); } /** * 查詢表中所有行(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)鍵字查詢報(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í)間戳范圍查詢報(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ù)類里注入hbaseUtils就可以使用了:
@Autowired
private HBaseUtils hBaseUtils;
補(bǔ)充知識(shí):springboot整合Hbase
springboot項(xiàng)目需要整合SpringCloud
依賴
<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多例模式 //若是在其他類中直接@Autowired引入的,多例就無(wú)效了,因?yàn)槟莻€(gè)類在初始化的時(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)線程安全的連接池 */ @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); } }
查詢
根據(jù)主鍵查詢內(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工具類)就是小編分享給大家的全部?jī)?nèi)容了,希望能給大家一個(gè)參考,也希望大家多多支持腳本之家。
相關(guān)文章
一文搞懂Mybatis中Mapper配置文件獲取參數(shù)的五種方式
這篇文章主要介紹了Mybatis中Mapper配置文件獲取參數(shù)的五種方式,文中通過(guò)代碼示例講解的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作有一定的幫助,需要的朋友可以參考下2024-03-03MyBatisPlus 自定義sql語(yǔ)句的實(shí)現(xiàn)
這篇文章主要介紹了MyBatisPlus 自定義sql語(yǔ)句的實(shí)現(xiàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-08-08使用MyBatis攔截器實(shí)現(xiàn)sql查詢權(quán)限動(dòng)態(tài)修改代碼實(shí)例
這篇文章主要介紹了使用MyBatis攔截器實(shí)現(xiàn)sql查詢權(quán)限動(dòng)態(tài)修改代碼實(shí)例,為了不耦合,現(xiàn)在的方案是在需要鑒權(quán)的Mybatis?Mapper方法上增加一個(gè)注解,在運(yùn)行過(guò)程中判斷該注解存在即對(duì)sql進(jìn)行修改,需要的朋友可以參考下2023-08-08詳解java中反射機(jī)制(含數(shù)組參數(shù))
這篇文章主要介紹了詳解java中反射機(jī)制(含數(shù)組參數(shù))的相關(guān)資料,希望通過(guò)本文能幫助到大家,讓大家理解掌握這部分內(nèi)容,需要的朋友可以參考下2017-10-10idea常用的18個(gè)設(shè)置(程序員必不可少)
這篇文章主要給大家介紹了關(guān)于idea常用的18個(gè)設(shè)置,這些對(duì)程序員們來(lái)說(shuō)必不可少,idea開(kāi)發(fā)常用基本且非常實(shí)用的配置,文中通過(guò)圖文介紹的非常詳細(xì),需要的朋友可以參考下2023-08-08