Java 操作 HBase 教程

---

---

相关博文原文地址：

博客园：美码师：HBase(2) Java 操作 HBase 教程

---

---

一、简介

HBase 本身就是用 Java 编写的，天生自带了 Java 原生API。 我们可以通过 hbase-client 来实现 HBase 数据库的操作。

所以，这次主要介绍该组件的基本用法。

在使用 hbase-client 之前，有几个要点需要注意：

• 客户端需要能访问 Zoopkeeper，再获得 HMaster、RegionServer 实例进行操作
• 客户端需运行在HBase/Hadoop 集群内，HBase会使用 hostname 来定位节点，因此要求客户端能访问到对应的主机名(或子域名)
  
  如果是远程客户端则需要配置本地的hosts文件。

下面这个图，有助于理解 Client 与 HBase 集群的交互架构：

二、hbase-client 引入

在 Maven 的 pom.xml 中添加依赖：

<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>2.1.5</version>
<exclusions>
<exclusion>
<artifactId>slf4j-api</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase</artifactId>
<version>2.1.5</version>
</dependency>


客户端版本和 HBase 版本需要保持一致，否则可能会遇到不兼容的问题。

三、连接操作

/**
* 建立连接
*
* @return
*/
public static Connection getConnection() {
try {
//获取配置
Configuration configuration = getConfiguration();
//检查配置
HBaseAdmin.checkHBaseAvailable(configuration);
return ConnectionFactory.createConnection(configuration);
} catch (IOException | ServiceException e) {
throw new RuntimeException(e);
}
}/**
* 获取配置
*
* @return
*/
private static Configuration getConfiguration() {
try {
Properties props = PropertiesLoaderUtils.loadAllProperties("hbase.properties");
String clientPort = props.getProperty("hbase.zookeeper.property.clientPort");
String quorum = props.getProperty("hbase.zookeeper.quorum"); logger.info("connect to zookeeper {}:{}", quorum, clientPort); Configuration config = HBaseConfiguration.create();
config.set("hbase.zookeeper.property.clientPort", clientPort);
config.set("hbase.zookeeper.quorum", quorum);
return config;
} catch (IOException e) {
throw new RuntimeException(e);
}
}


四、表操作

增删改查方法封装如下：

/**
* 创建表
* @param connection
* @param tableName
* @param columnFamilies
* @throws IOException
*/
public static void createTable(Connection connection, TableName tableName, String... columnFamilies) throws IOException {
Admin admin = null;
try {
admin = connection.getAdmin();
if (admin.tableExists(tableName)) {
logger.warn("table:{} exists!", tableName.getName());
} else {
TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);
for (String columnFamily : columnFamilies) {
builder.setColumnFamily(ColumnFamilyDescriptorBuilder.of(columnFamily));
}
admin.createTable(builder.build());
logger.info("create table:{} success!", tableName.getName());
}
} finally {
if (admin != null) {
admin.close();
}
}
}/**
* 插入数据
*
* @param connection
* @param tableName
* @param rowKey
* @param columnFamily
* @param column
* @param data
* @throws IOException
*/
public static void put(Connection connection, TableName tableName,
String rowKey, String columnFamily, String column, String data) throws IOException { Table table = null;
try {
table = connection.getTable(tableName);
Put put = new Put(Bytes.toBytes(rowKey));
put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(data));
table.put(put);
} finally {
if (table != null) {
table.close();
}
}
}/**
* 根据row key、column 读取
*
* @param connection
* @param tableName
* @param rowKey
* @param columnFamily
* @param column
* @throws IOException
*/
public static String getCell(Connection connection, TableName tableName, String rowKey, String columnFamily, String column) throws IOException {
Table table = null;
try {
table = connection.getTable(tableName);
Get get = new Get(Bytes.toBytes(rowKey));
get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column)); Result result = table.get(get);
List<Cell> cells = result.listCells(); if (CollectionUtils.isEmpty(cells)) {
return null;
}
String value = new String(CellUtil.cloneValue(cells.get(0)), "UTF-8");
return value;
} finally {
if (table != null) {
table.close();
}
}
}/**
* 根据rowkey 获取一行
*
* @param connection
* @param tableName
* @param rowKey
* @return
* @throws IOException
*/
public static Map<String, String> getRow(Connection connection, TableName tableName, String rowKey) throws IOException {
Table table = null;
try {
table = connection.getTable(tableName);
Get get = new Get(Bytes.toBytes(rowKey)); Result result = table.get(get);
List<Cell> cells = result.listCells(); if (CollectionUtils.isEmpty(cells)) {
return Collections.emptyMap();
}
Map<String, String> objectMap = new HashMap<>();
for (Cell cell : cells) {
String qualifier = new String(CellUtil.cloneQualifier(cell));
String value = new String(CellUtil.cloneValue(cell), "UTF-8");
objectMap.put(qualifier, value);
}
return objectMap;
} finally {
if (table != null) {
table.close();
}
}
}/**
* 扫描权标的内容
*
* @param connection
* @param tableName
* @param rowkeyStart
* @param rowkeyEnd
* @throws IOException
*/
public static List<Map<String, String>> scan(Connection connection, TableName tableName, String rowkeyStart, String rowkeyEnd) throws IOException {
Table table = null;
try {
table = connection.getTable(tableName);
ResultScanner rs = null;
try {
Scan scan = new Scan();
if (!StringUtils.isEmpty(rowkeyStart)) {
scan.withStartRow(Bytes.toBytes(rowkeyStart));
}
if (!StringUtils.isEmpty(rowkeyEnd)) {
scan.withStopRow(Bytes.toBytes(rowkeyEnd));
}
rs = table.getScanner(scan); List<Map<String, String>> dataList = new ArrayList<>();
for (Result r : rs) {
Map<String, String> objectMap = new HashMap<>();
for (Cell cell : r.listCells()) {
String qualifier = new String(CellUtil.cloneQualifier(cell));
String value = new String(CellUtil.cloneValue(cell), "UTF-8");
objectMap.put(qualifier, value);
}
dataList.add(objectMap);
}
return dataList;
} finally {
if (rs != null) {
rs.close();
}
}
} finally {
if (table != null) {
table.close();
}
}
}/**
* 删除表
*
* @param connection
* @param tableName
* @throws IOException
*/
public static void deleteTable(Connection connection, TableName tableName) throws IOException {
Admin admin = null;
try {
admin = connection.getAdmin();
if (admin.tableExists(tableName)) {
//现执行disable
admin.disableTable(tableName);
admin.deleteTable(tableName);
}
} finally {
if (admin != null) {
admin.close();
}
}
}


五、运行测试

1. 建立 DeviceState 表；
2. 定义 name/state 两个列簇；
3. 写入列数据；
4. 读取列、行，范围读取；
5. 删除操作

private static final Logger logger = LoggerFactory.getLogger(HBaseTest.class);public static void main(String[] args) { Connection connection = null;
try {
connection = getConnection();
TableName tableName = TableName.valueOf("DeviceState"); //创建DeviceState表
createTable(connection, tableName, "name", "state"); logger.info("创建表 {}", tableName.getNameAsString()); //写入数据
put(connection, tableName, "row1", "name", "c1", "空调");
put(connection, tableName, "row1", "state", "c2", "打开");
put(connection, tableName, "row2", "name", "c1", "电视机");
put(connection, tableName, "row2", "state", "c2", "关闭"); logger.info("写入数据."); String value = getCell(connection, tableName, "row1", "state", "c2");
logger.info("读取单元格-row1.state:{}", value); Map<String, String> row = getRow(connection, tableName, "row2");
logger.info("读取单元格-row2:{}", JsonUtil.toJson(row)); List<Map<String, String>> dataList = scan(connection, tableName, null, null);
logger.info("扫描表结果-:\n{}", JsonUtil.toPrettyJson(dataList)); //删除DeviceState表
deleteTable(connection, tableName);
logger.info("删除表 {}", tableName.getNameAsString()); logger.info("操作完成.");
} catch (Exception e) {
logger.error("操作出错", e);
} finally {
if (connection != null) {
try {
connection.close();
} catch (IOException e) {
logger.error("error occurs", e);
}
}
}}


执行代码，控制台输出如下：

INFO -createTable(HBaseTest.java:89) - create table:[68, 101, 118, 105, 99, 101, 83, 116, 97, 116, 101] success!
INFO -main(HBaseTest.java:32) - 创建表 DeviceState
INFO -main(HBaseTest.java:40) - 写入数据.
INFO -main(HBaseTest.java:43) - 读取单元格-row1.state:打开
INFO -main(HBaseTest.java:46) - 读取单元格-row2:{"c1":"电视机","c2":"关闭"}
INFO -main(HBaseTest.java:49) - 扫描表结果-:
[ {
"c1" : "空调",
"c2" : "打开"
}, {
"c1" : "电视机",
"c2" : "关闭"
} ]
INFO -HBaseAdmin$9.call(HBaseAdmin.java:1380) - Started disable of DeviceState
INFO -HBaseAdmin$DisableTableFuture.postOperationResult(HBaseAdmin.java:1409) - Disabled DeviceState
INFO -HBaseAdmin$DeleteTableFuture.postOperationResult(HBaseAdmin.java:965) - Deleted DeviceState
INFO -main(HBaseTest.java:53) - 删除表 DeviceState
INFO -main(HBaseTest.java:55) - 操作完成.