Hadoop 之 Hbase 配置与使用(四)
•
大数据
Hadoop 之 Hbase 配置与使用
- 一.Hbase 下载
-
- 1.Hbase 下载
- 二.Hbase 配置
-
- 1.单机部署
- 2.伪集群部署(基于单机配置)
- 3.集群部署
-
- 1.启动 hadoop 集群
- 2.启动 zookeeper 集群
- 3.启动 hbase 集群
- 4.集群启停脚本
- 三.测试
-
- 1.Pom 配置
- 2.Yml 配置
- 3.Hbase 配置类
- 4.Hbase 连接池配置
- 5.测试类
- 6.启动类
- 7.测试
一.Hbase 下载
HBase 是一个分布式的、面向列的开源数据库:Hbase API
1.Hbase 下载
Hbase 下载

跳转到下载链接

二.Hbase 配置
1.单机部署
## 1.创建安装目录
mkdir -p /usr/local/hbase
## 2.将压缩包拷贝到虚拟机并解压缩
tar zxvf hbase-3.0.0-alpha-4-bin.tar.gz -C /usr/local/hbase/
## 3.添加环境变量
echo 'export HBASE_HOME=/usr/local/hbase/hbase-3.0.0-alpha-4' >> /etc/profile
echo 'export PATH=${HBASE_HOME}/bin:${PATH}' >> /etc/profile
source /etc/profile
## 4.指定 JDK 版本
echo 'export JAVA_HOME=/usr/local/java/jdk-11.0.19' >> $HBASE_HOME/conf/hbase-env.sh
## 5.创建 hbase 存储目录
mkdir -p /home/hbase/data
## 6.修改配置
vim $HBASE_HOME/conf/hbase-site.xml
添加如下信息
hbase.rootdir
file:///home/hbase/data

## 1.进入安装目录 cd $HBASE_HOME ## 2.启动服务 ./bin/start-hbase.sh

## 1.进入安装目录 cd $HBASE_HOME ## 2.关闭服务 ./bin/stop-hbase.sh

2.伪集群部署(基于单机配置)
## 1.修改 hbase-env.sh echo 'export JAVA_HOME=/usr/local/java/jdk-11.0.19' >> $HBASE_HOME/conf/hbase-env.sh echo 'export HBASE_MANAGES_ZK=true' >> $HBASE_HOME/conf/hbase-env.sh ## 2.修改 hbase_site.xml vim $HBASE_HOME/conf/hbase-site.xml hbase.rootdir hdfs://nn/hbase hbase.cluster.distributed true hbase.zookeeper.quorum nn dfs.replication 1 ## 3.修改 regionservers 的 localhost 为 nn echo nn > $HBASE_HOME/conf/regionservers

## 1.进入安装目录 cd $HADOOP_HOME ## 2.启动 hadoop 服务 ./sbin/start-all.sh

## 1.进入安装目录 cd $HBASE_HOME ## 2.启动服务 ./bin/start-hbase.sh

## 1.进入安装目录 cd $HBASE_HOME ## 2.关闭主节点服务(直接关服务是关不掉的,如图) . bin/hbase-daemon.sh stop master ## 3.关闭服务 ./bin/stop-hbase.sh

3.集群部署
## 1.创建 zookeeper 数据目录
mkdir -p $HBASE_HOME/zookeeper/data
## 2.进入安装目录
cd $HBASE_HOME/conf
## 3.修改环境配置
vim hbase-env.sh
## 添加 JDK / 启动外置 Zookeeper
# JDK
export JAVA_HOME=/usr/local/java/jdk-11.0.19
# Disable Zookeeper
export HBASE_MANAGES_ZK=false
## 4.修改 hbase-site.xml
vim hbase-site.xml
## 配置如下信息
hbase.master.maxclockskew`
6000
hbase.rootdir
hdfs://nn:9000/hbase
hbase.cluster.distributed
true
hbase.zookeeper.quorum
zk1,zk2,zk3
hbase.zookeeper.property.dataDir
/usr/local/hbase/hbase-3.0.0-alpha-4/zookeeper/data
hbase.wal.provider
filesystem
## 5.清空 regionservers 并添加集群节点域名
echo '' > regionservers
echo 'nn' >> regionservers
echo 'nd1' >> regionservers
echo 'nd2' >> regionservers
## 6.分别为 nd1 / nd2 创建 hbase 目录
mkdir -p /usr/local/hbase
## 7.分发 hbase 配置到另外两台虚拟机 nd1 / nd2
scp -r /usr/local/hbase/hbase-3.0.0-alpha-4 root@nd1:/usr/local/hbase
scp -r /usr/local/hbase/hbase-3.0.0-alpha-4 root@nd2:/usr/local/hbase
## 8.分发环境变量配置
scp /etc/profile root@nd1:/etc/profile
scp /etc/profile root@nd2:/etc/profile
1.启动 hadoop 集群
Hadoop 集群搭建参考:Hadoop 搭建
## 1.启动 hadoop cd $HADOOP_HOME . sbin/start-all.sh

## 1.关闭 hadoop 安全模式 hadoop dfsadmin -safemode leave

2.启动 zookeeper 集群
ZOOKEEPER 集群搭建说明
## 1.启动 zookeeper 集群 zkServer.sh start && ssh root@zk2 "source /etc/profile && zkServer.sh start && exit" && ssh root@zk3 "source /etc/profile && zkServer.sh start && exit" ## 2.查看状态 zkServer.sh status && ssh root@zk2 "source /etc/profile && zkServer.sh status && exit" && ssh root@zk3 "source /etc/profile && zkServer.sh status && exit"

3.启动 hbase 集群
## 1.分别为 nn /nd1 / nd2 配置 zookeeper 域名解析 echo '192.168.1.100 zk1' >> /etc/hosts echo '192.168.1.101 zk2' >> /etc/hosts echo '192.168.1.102 zk3' >> /etc/hosts ## 2.启动 habase cd $HBASE_HOME . bin/start-hbase.sh ## 3.停止服务 . bin/hbase-daemon.sh stop master . bin/hbase-daemon.sh stop regionserver . bin/stop-hbase.sh

查看 UI 监控:http://192.168.1.6:16010/master-status

4.集群启停脚本
#!/bin/bash case $1 in "start") ## start hadoop start-all.sh ## start zookeeper (先配置免密登录) zkServer.sh start && ssh root@zk2 "source /etc/profile && zkServer.sh start && exit" && ssh root@zk3 "source /etc/profile && zkServer.sh start && exit" ## start hbase start-hbase.sh ;; "stop") ## stop hbase ssh root@nd1 "source /etc/profile && hbase-daemon.sh stop regionserver && stop-hbase.sh && exit" ssh root@nd2 "source /etc/profile && hbase-daemon.sh stop regionserver && stop-hbase.sh && exit" hbase-daemon.sh stop master && hbase-daemon.sh stop regionserver && stop-hbase.sh ## stop zookeeper zkServer.sh stop && ssh root@zk2 "source /etc/profile && zkServer.sh stop && exit" && ssh root@zk3 "source /etc/profile && zkServer.sh stop && exit" ## stop hadoop stop-all.sh ;; *) echo "pls inout start|stop" ;; esac
三.测试
## 1.为 Windows 增加 Hosts 配置,添加 Hbase 集群域名解析 编辑如下文件 C:\Windows\System32\drivers\etc\hosts ## 2.增加如下信息 192.168.1.6 nn 192.168.1.7 nd1 192.168.1.8 nd2
测试配置效果

JDK 版本

工程结构

1.Pom 配置
4.0.0
org.example
hbase-demo
1.0-SNAPSHOT
11
11
2.7.8
UTF-8
org.springframework.boot
spring-boot-starter-web
${spring.version}
org.projectlombok
lombok
1.18.28
com.alibaba
fastjson
2.0.32
org.apache.hbase
hbase-client
3.0.0-alpha-4
2.Yml 配置
hbase:
zookeeper:
quorum: 192.168.1.100,192.168.1.101,192.168.1.102
property:
clientPort: 2181
master:
ip: 192.168.1.6
port: 16000
3.Hbase 配置类
package org.example.config;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author Administrator
* @Description
* @create 2023-07-25 0:26
*/
@Configuration
public class HbaseConfig {
@Value("${hbase.zookeeper.quorum}")
private String zookeeperQuorum;
@Value("${hbase.zookeeper.property.clientPort}")
private String clientPort;
@Value("${hbase.master.ip}")
private String ip;
@Value("${hbase.master.port}")
private int masterPort;
@Bean
public org.apache.hadoop.conf.Configuration hbaseConfiguration(){
org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum",zookeeperQuorum);
conf.set("hbase.zookeeper.property.clientPort",clientPort);
conf.set("hbase.masters", ip + ":" + masterPort);
conf.set("hbase.client.keyvalue.maxsize","20971520");
return HBaseConfiguration.create(conf);
}
}
4.Hbase 连接池配置
package org.example.config;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Enumeration;
import java.util.Vector;
/**
* @author Administrator
* @Description
* @create 2023-07-25 22:39
*/
@Slf4j
@Component
public class HbaseConnectionPool {
/**
* 连接池的初始大小
* 连接池的创建步长
* 连接池最大的大小
*/
private int nInitConnectionAmount = 3;
private int nIncrConnectionAmount = 3;
private int nMaxConnections = 20;
/**
* 存放连接池中数据库连接的向量
*/
private Vector vcConnections = new Vector();
/**
* 注入连接配置
*/
@Resource
private Configuration hbaseConfiguration;
/**
* 初始化连接
*/
@PostConstruct
public void init() {
createConnections(nInitConnectionAmount);
}
/**
* 获取可用连接
* @return
*/
public synchronized Connection getConnection() {
Connection conn;
while (null == (conn =getFreeConnection())){
try {
wait(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 返回获得的可用的连接
return conn;
}
/**
* 释放连接
* @param conn
*/
public synchronized void releaseConnection(Connection conn) {
ConnectionWrapper connWrapper;
Enumeration enumerate = this.vcConnections.elements();
while(enumerate.hasMoreElements()) {
connWrapper = (ConnectionWrapper) enumerate.nextElement();
if (conn == connWrapper.getConnection()) {
connWrapper.setBusy(false);
break;
}
}
}
/**
* 获取可用连接 当前无可用连接则创建 如果已达到最大连接数则返回 null 阻塞后重试获取
* @return
*/
private Connection getFreeConnection() {
Connection conn;
if (null == (conn = findFreeConnection())) {
// 创建新连接
createConnections(nIncrConnectionAmount);
// 查看是否有可用连接
if (null == (conn = findFreeConnection())) {
return null;
}
}
return conn;
}
/**
* 查找可用连接
* @return
*/
private Connection findFreeConnection() {
ConnectionWrapper connWrapper;
//遍历向量内连接对象
Enumeration enumerate = vcConnections.elements();
while (enumerate.hasMoreElements()) {
connWrapper = (ConnectionWrapper) enumerate.nextElement();
//判断当前连接是否被占用
if (!connWrapper.isBusy()) {
connWrapper.setBusy(true);
return connWrapper.getConnection();
}
}
// 返回 NULL
return null;
}
/**
* 创建新连接
* @param counts
*/
private void createConnections(int counts) {
// 循环创建指定数目的数据库连接
try {
for (int i = 0; i < counts; i++) {
if (this.nMaxConnections > 0 && this.vcConnections.size() >= this.nMaxConnections) {
log.warn("已达到最大连接数...");
break;
}
// 创建一个新连接并加到向量
vcConnections.addElement(new ConnectionWrapper(newConnection()));
}
} catch (Exception e) {
log.error("创建连接失败...");
}
}
/**
* 创建新连接
* @return
*/
private Connection newConnection() {
/** hbase 连接 */
Connection conn = null;
// 创建一个数据库连接
try {
conn = ConnectionFactory.createConnection(hbaseConfiguration);
} catch (Exception e) {
log.error("HBase 连接失败...");
}
// 返回创建的新的数据库连接
return conn;
}
/**
* 封装连接对象
*/
@Data
class ConnectionWrapper {
/**
* 数据库连接
*/
private Connection connection;
/**
* 此连接是否正在使用的标志,默认没有正在使用
*/
private boolean busy = false;
/**
* 构造函数,根据一个 Connection 构告一个 PooledConnection 对象
*/
public ConnectionWrapper(Connection connection) {
this.connection = connection;
}
}
}
5.测试类
package org.example.controller;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.util.Bytes;
import org.example.config.HbaseConnectionPool;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
/**
* @author Administrator
*
* 可利用 aop 进行连接获取和释放处理
*
* @Description
* @create 2023-07-25 23:06
*/
@Slf4j
@RestController
@RequestMapping("/hbase")
public class HbaseController {
@Resource
private HbaseConnectionPool pool;
/**
* 表名
*/
private String tbl_user = "tbl_user";
/**
* 创建表(不允许重复创建)
*/
@GetMapping("/create")
public void createTable(){
Connection conn = null;
//获取连接
try {
conn = pool.getConnection();
Admin admin = conn.getAdmin();
TableName tableName = TableName.valueOf(tbl_user);
if (!admin.tableExists(tableName)){
//指定表名
TableDescriptorBuilder tdb_user = TableDescriptorBuilder.newBuilder(tableName);
//添加列族(info,data)
ColumnFamilyDescriptor hcd_info = ColumnFamilyDescriptorBuilder.of("name");
ColumnFamilyDescriptor hcd_data = ColumnFamilyDescriptorBuilder.of("age");
tdb_user.setColumnFamily(hcd_info);
tdb_user.setColumnFamily(hcd_data);
//创建表
TableDescriptor td = tdb_user.build();
admin.createTable(td);
}
} catch (IOException e) {
throw new RuntimeException(e);
} finally {
if (null != conn){
pool.releaseConnection(conn);
}
}
}
/**
* 删除表(不允许删除不存在的表)
*/
@GetMapping("/drop")
public void dropTable(){
Connection conn = null;
try {
conn = pool.getConnection();
Admin admin = conn.getAdmin();
TableName tableName = TableName.valueOf(tbl_user);
if (admin.tableExists(tableName)){
admin.disableTable(tableName);
admin.deleteTable(tableName);
}
} catch (IOException e) {
throw new RuntimeException(e);
} finally {
if (null != conn){
pool.releaseConnection(conn);
}
}
}
/**
* 插入测试
*/
@GetMapping("/insert")
public void insert(){
log.info("---插入一列数据---1");
putData(tbl_user, "row1", "name", "a", "zhangSan");
putData(tbl_user, "row1", "age", "a", "18");
log.info("---插入多列数据---2");
putData(tbl_user, "row2", "name",
Arrays.asList("a", "b", "c"), Arrays.asList("liSi", "wangWu", "zhaoLiu"));
log.info("---插入多列数据---3");
putData(tbl_user, "row3", "age",
Arrays.asList("a", "b", "c"), Arrays.asList("18","19","20"));
log.info("---插入多列数据---4");
putData(tbl_user, "row4", "age",
Arrays.asList("a", "b", "c"), Arrays.asList("30","19","20"));
}
/**
* 插入数据(单条)
* @param tableName 表名
* @param rowKey rowKey
* @param columnFamily 列族
* @param column 列
* @param value 值
* @return true/false
*/
public boolean putData(String tableName, String rowKey, String columnFamily, String column,
String value) {
return putData(tableName, rowKey, columnFamily, Arrays.asList(column),
Arrays.asList(value));
}
/**
* 插入数据(批量)
* @param tableName 表名
* @param rowKey rowKey
* @param columnFamily 列族
* @param columns 列
* @param values 值
* @return true/false
*/
public boolean putData(String tableName, String rowKey, String columnFamily,
List columns, List values) {
Connection conn = null;
try {
conn = pool.getConnection();
Table table = conn.getTable(TableName.valueOf(tableName));
Put put = new Put(Bytes.toBytes(rowKey));
for (int i=0; i<columns.size(); i++) {
put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columns.get(i)), Bytes.toBytes(values.get(i)));
}
table.put(put);
table.close();
return true;
} catch (IOException e) {
e.printStackTrace();
return false;
} finally {
if (null != conn){
pool.releaseConnection(conn);
}
}
}
/**
* 查询测试
*/
@GetMapping("/query")
public void getResultScanner(){
log.info("全表数据:{}",getData(tbl_user));
log.info("过滤器,按年龄 [18]:{}",getData(tbl_user,new ColumnValueFilter(Bytes.toBytes("age"), Bytes.toBytes("a"), CompareOperator.EQUAL, Bytes.toBytes("18"))));
log.info("根据 rowKey [row1]:{}",getData(tbl_user,"row1"));
log.info("根据 rowKey 列族 列 [row2 name a]:{}",getData(tbl_user,"row2","name","a"));
}
/**
* 获取数据(全表数据)
* @param tableName 表名
* @return map
*/
public List<Map> getData(String tableName) {
List<Map> list = new ArrayList();
Connection conn = null;
try {
conn = pool.getConnection();
Table table = conn.getTable(TableName.valueOf(tableName));
Scan scan = new Scan();
ResultScanner resultScanner = table.getScanner(scan);
for(Result result : resultScanner) {
HashMap map = new HashMap(result.listCells().size());
map.put("row", Bytes.toString(result.getRow()));
for (Cell cell : result.listCells()) {
//列族
String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
//列
String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
//值
String data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
map.put(family + ":" + qualifier, data);
}
list.add(map);
}
table.close();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (null != conn){
pool.releaseConnection(conn);
}
}
return list;
}
/**
* 获取数据(根据 filter)
* @param tableName 表名
* @param filter 过滤器
* @return map
*/
public List<Map> getData(String tableName, Filter filter) {
List<Map> list = new ArrayList();
Connection conn = null;
try {
conn = pool.getConnection();
Table table = conn.getTable(TableName.valueOf(tableName));
Scan scan = new Scan();
// 添加过滤器
scan.setFilter(filter);
ResultScanner resultScanner = table.getScanner(scan);
for(Result result : resultScanner) {
HashMap map = new HashMap(result.listCells().size());
map.put("row", Bytes.toString(result.getRow()));
for (Cell cell : result.listCells()) {
String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
String data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
map.put(family + ":" + qualifier, data);
}
list.add(map);
}
table.close();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (null != conn){
pool.releaseConnection(conn);
}
}
return list;
}
/**
* 获取数据(根据 rowKey)
* @param tableName 表名
* @param rowKey rowKey
* @return map
*/
public Map getData(String tableName, String rowKey) {
HashMap map = new HashMap();
Connection conn = null;
try {
conn = pool.getConnection();
Table table = conn.getTable(TableName.valueOf(tableName));
Get get = new Get(Bytes.toBytes(rowKey));
Result result = table.get(get);
if (result != null && !result.isEmpty()) {
for (Cell cell : result.listCells()) {
String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
String data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
map.put(family + ":" + qualifier, data);
}
}
table.close();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (null != conn){
pool.releaseConnection(conn);
}
}
return map;
}
/**
* 获取数据(根据 rowKey 列族 列)
* @param tableName 表名
* @param rowKey rowKey
* @param columnFamily 列族
* @param columnQualifier 列
* @return map
*/
public String getData(String tableName, String rowKey, String columnFamily,
String columnQualifier) {
String data = "";
Connection conn = null;
try {
conn = pool.getConnection();
Table table = conn.getTable(TableName.valueOf(tableName));
Get get = new Get(Bytes.toBytes(rowKey));
get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnQualifier));
Result result = table.get(get);
if (result != null && !result.isEmpty()) {
Cell cell = result.listCells().get(0);
data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
}
table.close();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (null != conn){
pool.releaseConnection(conn);
}
}
return data;
}
/**
* 删除数据
*/
@GetMapping("/delete")
public void delete(){
log.info("---删除 rowKey --- row1 ");
deleteData(tbl_user,"row1");
log.info("---删除 rowKey 列族 --- row2 age ");
deleteData(tbl_user,"row2","age");
}
/**
* 删除数据(根据 rowKey)
* @param tableName 表名
* @param rowKey rowKey
*/
public void deleteData(String tableName, String rowKey) {
Connection conn = null;
try {
conn = pool.getConnection();
Table table = conn.getTable(TableName.valueOf(tableName));
Delete delete = new Delete(Bytes.toBytes(rowKey));
table.delete(delete);
table.close();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (null != conn){
pool.releaseConnection(conn);
}
}
}
/**
* 删除数据(根据 row key,列族)
* @param tableName 表名
* @param rowKey rowKey
* @param columnFamily 列族
*/
public void deleteData(String tableName, String rowKey, String columnFamily) {
Connection conn = null;
try {
conn = pool.getConnection();
Table table = conn.getTable(TableName.valueOf(tableName));
Delete delete = new Delete(Bytes.toBytes(rowKey));
delete.addFamily(columnFamily.getBytes());
table.delete(delete);
table.close();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (null != conn){
pool.releaseConnection(conn);
}
}
}
}
6.启动类
package org.example;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author Administrator
*/
@SpringBootApplication
public class HbaseDemo {
public static void main(String[] args) {
SpringApplication.run(HbaseDemo.class,args);
}
}
7.测试
创建表:http://127.0.0.1:8080/hbase/create 插入:http://127.0.0.1:8080/hbase/insert 查询:http://127.0.0.1:8080/hbase/query 删除:http://127.0.0.1:8080/hbase/delete 删除表:http://127.0.0.1:8080/hbase/drop
查看 UI

测试输出日志

本文来自网络,不代表协通编程立场,如若转载,请注明出处:https://net2asp.com/fd9f84237c.html
