SpringBoot整合Hbase的實現示例
簡介
當單表數據量過大的時候，關系性數據庫會出現性能瓶頸，這時候我們就可以用NoSql，比如Hbase就是一個不錯的解決方案。接下來是用Spring整合Hbase的實際案例，且在最後會給出整合中可能會出現的問題，以及解決方案。這裡我是用本地Windows的IDEA，與局域網的偽分佈Hbase集群做的連接，其中Hbase集群包括的組件有：Jdk1.8、Hadoop2.7.6、ZooKeeper3.4.10、Hbase2.0.1，因為這裡隻是開發環境，所以做一個偽分佈的就好，之後部署的時候再按生產環境要求來即可
整合步驟
目錄結構
pom.xml
這裡要導入Hbase連接所需要包，需要找和你Hbase版本一致的包
<dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>2.0.1</version> </dependency>
hbase-site.xml
我是用的配置文件連接方法，這個配置文件你在hbase的安裝目錄下的conf目錄就可以找到，然後你直接把它復制到項目的resources目錄下就好，當然你也可以用application.properties配置文件外加註入和代碼的方式代替這個配置文件
HBaseConfig.java
這裡因為隻需連接Hbase就沒連接Hadoop，如果要連接Hadoop，Windows下還要下載winutils.exe工具，後面會介紹
@Configuration
public class HBaseConfig {
@Bean
public HBaseService getHbaseService() {
//設置臨時的hadoop環境變量，之後程序會去這個目錄下的\bin目錄下找winutils.exe工具，windows連接hadoop時會用到
//System.setProperty("hadoop.home.dir", "D:\\Program Files\\Hadoop");
//執行此步時，會去resources目錄下找相應的配置文件，例如hbase-site.xml
org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
return new HBaseService(conf);
}
}
HBaseService.java
這是做連接後的一些操作可以參考之後自己寫一下
public class HBaseService {
private Logger log = LoggerFactory.getLogger(HBaseService.class);
/**
* 管理員可以做表以及數據的增刪改查功能
*/
private Admin admin = null;
private Connection connection = null;
public HBaseService(Configuration conf) {
try {
connection = ConnectionFactory.createConnection(conf);
admin = connection.getAdmin();
} catch (IOException e) {
log.error("獲取HBase連接失敗!");
}
}
/**
* 創建表 create <table>, {NAME => <column family>, VERSIONS => <VERSIONS>}
*/
public boolean creatTable(String tableName, List<String> columnFamily) {
try {
//列族column family
List<ColumnFamilyDescriptor> cfDesc = new ArrayList<>(columnFamily.size());
columnFamily.forEach(cf -> {
cfDesc.add(ColumnFamilyDescriptorBuilder.newBuilder(
Bytes.toBytes(cf)).build());
});
//表 table
TableDescriptor tableDesc = TableDescriptorBuilder
.newBuilder(TableName.valueOf(tableName))
.setColumnFamilies(cfDesc).build();
if (admin.tableExists(TableName.valueOf(tableName))) {
log.debug("table Exists!");
} else {
admin.createTable(tableDesc);
log.debug("create table Success!");
}
} catch (IOException e) {
log.error(MessageFormat.format("創建表{0}失敗", tableName), e);
return false;
} finally {
close(admin, null, null);
}
return true;
}
/**
* 查詢所有表的表名
*/
public List<String> getAllTableNames() {
List<String> result = new ArrayList<>();
try {
TableName[] tableNames = admin.listTableNames();
for (TableName tableName : tableNames) {
result.add(tableName.getNameAsString());
}
} catch (IOException e) {
log.error("獲取所有表的表名失敗", e);
} finally {
close(admin, null, null);
}
return result;
}
/**
* 遍歷查詢指定表中的所有數據
*/
public Map<String, Map<String, String>> getResultScanner(String tableName) {
Scan scan = new Scan();
return this.queryData(tableName, scan);
}
/**
* 通過表名及過濾條件查詢數據
*/
private Map<String, Map<String, String>> queryData(String tableName, Scan scan) {
// <rowKey,對應的行數據>
Map<String, Map<String, String>> result = new HashMap<>();
ResultScanner rs = null;
//獲取表
Table table = null;
try {
table = getTable(tableName);
rs = table.getScanner(scan);
for (Result r : rs) {
// 每一行數據
Map<String, String> columnMap = new HashMap<>();
String rowKey = null;
// 行鍵，列族和列限定符一起確定一個單元（Cell）
for (Cell cell : r.listCells()) {
if (rowKey == null) {
rowKey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
}
columnMap.put(
//列限定符
Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()),
//列族
Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
}
if (rowKey != null) {
result.put(rowKey, columnMap);
}
}
} catch (IOException e) {
log.error(MessageFormat.format("遍歷查詢指定表中的所有數據失敗,tableName:{0}", tableName), e);
} finally {
close(null, rs, table);
}
return result;
}
/**
* 為表添加或者更新數據
*/
public void putData(String tableName, String rowKey, String familyName, String[] columns, String[] values) {
Table table = null;
try {
table = getTable(tableName);
putData(table, rowKey, tableName, familyName, columns, values);
} catch (Exception e) {
log.error(MessageFormat.format("為表添加 or 更新數據失敗,tableName:{0},rowKey:{1},familyName:{2}", tableName, rowKey, familyName), e);
} finally {
close(null, null, table);
}
}
private void putData(Table table, String rowKey, String tableName, String familyName, String[] columns, String[] values) {
try {
//設置rowkey
Put put = new Put(Bytes.toBytes(rowKey));
if (columns != null && values != null && columns.length == values.length) {
for (int i = 0; i < columns.length; i++) {
if (columns[i] != null && values[i] != null) {
put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columns[i]), Bytes.toBytes(values[i]));
} else {
throw new NullPointerException(MessageFormat.format(
"列名和列數據都不能為空,column:{0},value:{1}", columns[i], values[i]));
}
}
}
table.put(put);
log.debug("putData add or update data Success,rowKey:" + rowKey);
table.close();
} catch (Exception e) {
log.error(MessageFormat.format(
"為表添加 or 更新數據失敗,tableName:{0},rowKey:{1},familyName:{2}",
tableName, rowKey, familyName), e);
}
}
/**
* 根據表名獲取table
*/
private Table getTable(String tableName) throws IOException {
return connection.getTable(TableName.valueOf(tableName));
}
/**
* 關閉流
*/
private void close(Admin admin, ResultScanner rs, Table table) {
if (admin != null) {
try {
admin.close();
} catch (IOException e) {
log.error("關閉Admin失敗", e);
}
if (rs != null) {
rs.close();
}
if (table != null) {
rs.close();
}
if (table != null) {
try {
table.close();
} catch (IOException e) {
log.error("關閉Table失敗", e);
}
}
}
}
}
HBaseApplicationTests.java
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
class HBaseApplicationTests {
@Resource
private HBaseService hbaseService;
//測試創建表
@Test
public void testCreateTable() {
hbaseService.creatTable("test_base", Arrays.asList("a", "back"));
}
//測試加入數據
@Test
public void testPutData() {
hbaseService.putData("test_base", "000001", "a", new String[]{
"project_id", "varName", "coefs", "pvalues", "tvalues",
"create_time"}, new String[]{"40866", "mob_3", "0.9416",
"0.0000", "12.2293", "null"});
hbaseService.putData("test_base", "000002", "a", new String[]{
"project_id", "varName", "coefs", "pvalues", "tvalues",
"create_time"}, new String[]{"40866", "idno_prov", "0.9317",
"0.0000", "9.8679", "null"});
hbaseService.putData("test_base", "000003", "a", new String[]{
"project_id", "varName", "coefs", "pvalues", "tvalues",
"create_time"}, new String[]{"40866", "education", "0.8984",
"0.0000", "25.5649", "null"});
}
//測試遍歷全表
@Test
public void testGetResultScanner() {
Map<String, Map<String, String>> result2 = hbaseService.getResultScanner("test_base");
System.out.println("-----遍歷查詢全表內容-----");
result2.forEach((k, value) -> {
System.out.println(k + "--->" + value);
});
}
}
運行結果
Hbase數據庫查詢結果
IDEA的遍歷結果
報錯與解決方案
報錯一
解決方案：
這是參數配置的有問題，如果你是用hbase-site.xml配置文件配置的參數，那麼檢查它，用代碼配置就檢查代碼參數
報錯二
解決方案：
更改windows本地hosts文件，C:\Windows\System32\drivers\etc\hosts,添加Hbase服務所在主機地址與主機名稱，這裡你如果保存不瞭hosts文件，把它拉出到桌面改好再拉回即可
報錯三
解決方案：
這是因為在Windows下連接Hadoop需要一個叫Winutils.exe的工具，並且從源代碼可知，它會去讀你Windows下的環境變量，如果你不想在本地設置，可以用方法System.setProperty()設置實時環境變量，另外，如果你隻用Hbase，其實這個報錯並不影響你使用Hbase服務
代碼地址
https://github.com/xiaoxiamo/SpringBoot_HBase
到此這篇關於SpringBoot整合Hbase的實現示例的文章就介紹到這瞭,更多相關SpringBoot整合Hbase內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- 詳解HBase表的數據模型
- Hbase列式存儲入門教程
- Linux下Hbase安裝配置教程
- Android動態表格的實現代碼(內容、樣式可擴縮)
- 通過DBeaver連接Phoenix操作hbase的方法