HBaseAPI 客户端使用介绍 - OceanBase 数据库 V4.1.0

本文从 OceanBase 的 HBaseAPI 使用角度出发，引导您正确使用 HBaseAPI 客户端。包括配置客户端、连接 OBserver，以及进行基本的增删改查操作。

背景信息

HBaseAPI 客户端基于 TableAPI 提供的基本接口，在客户端封装了 HBase 兼容的 API，目前已兼容 HBase 0.94 版本的特性。进行客户端操作前，需要了解下列 OceanBase 数据库的基本知识：

HBaseAPI 部署

添加客户端依赖

添加 HBaseAPI 客户端 jar 包依赖到本地 java 工程的 pom.xml 文件（或参考 HBaseAPI 使用 Demo）。

<dependency>
    <groupId>com.oceanbase</groupId>
    <artifactId>obkv-hbase-client</artifactId>
    <version>0.1.0</version>
</dependency>

<dependency>
    <groupId>com.oceanbase</groupId>
    <artifactId>obkv-table-client</artifactId>
    <version>0.1.0</version>
 </dependency>

建立 OB HBase 数据表

使用 OBClient 或 MySQL 连接 OBserver 集群，并建立 HBase 相关数据表，请参考 HBaseAPI 存储模型。

客户端接入初始化

获取 ConfigURL

从 OCP 部署的集群获取

1.登录 OceanBase 云平台。 2.在左侧导航栏选择集群，下滑找到 集群列表。 3.单击打开需要访问的集群名称。 4.在集群详细信息页找到 ConfigURL 参数，该参数在客户端初始化中需要配置。获取 ConfigURL

设置客户端对 OBServer 节点的登录用户&密码

使用 OBClient 或 MySQL 客户端连接 OBServer 集群，创建 SYS_USER_NAME 以及 SYS_PASSWORD。创建方式，请参考如何创建租户用户。下文示例通过客户端自带 proxyro 用户说明，但要设置 proxyro 密码，设置方法如下：

GRANT SELECT ON oceanbase.* TO proxyro IDENTIFIED BY 'your password';

客户端参数设置

下文示例描述了客户端参数设置的两种方式，如果您想获取更多的关于客户端参数配置选项信息，请参考：Property.java。由于 OHTable 的设计是非线程安全的模式，多线程并发使用 OHTable 时，建议使用 OHTablePool 申请 OHTable。

使用 OHTable 设置

import static com.alipay.oceanbase.rpc.property.Property.*

public class OHTableTest extends HTableTestBase {
    public void test() throws IOException {
        Configuration conf = new Configuration();
        // <必须参数>
        conf.set("hbase.oceanbase.paramURL","your paramURL"); 
        // 设置 configRUL
        // 设置登录用户名，例如 root@sys#obcluster，其中 root 为用户名，sys 为租户名，obcluster 为集群名
        conf.set("hbase.oceanbase.fullUserName", "your fullUserName"); 
        // 设置登录密码，如果没有设置登录密码可以忽略
        conf.set("hbase.oceanbase.password", "your login passwd"); 
        // 设置系统用户名，格式：用户@租户，例如，observer 集群自带的 proxyro 系统用户：proxyro@sys
        conf.set("hbase.oceanbase.sysUserName", "your sysUserName"); 
        // 设置系统用户登录密码。
        conf.set("hbase.oceanbase.sysPassword", "your sysPassword");
        // <可选参数>
        c.set("rs.list.acquire.read.timeout", "10000"); 
        // 获取 RS 列表的读取的超时时间
        c.set("rs.list.acquire.connect.timeout", "10000"); 
        // 获取 RS 列表的建连的超时时间
        c.set("rpc.connect.timeout", "1000"); // 建立 RPC 连接的超时时间
        c.set("rpc.execute.timeout", "1000"); // 执行 RPC 请求的 socket 超时时间
        
        hTable = new OHTable(conf, "your tablename");
    }
}

使用 OHTablePool 配置

import static com.alipay.oceanbase.rpc.property.Property.*;
import org.apache.hadoop.hbase.util.PoolMap;

public class OHTablePoolTest extends HTableTestBase {

    private OHTablePool ohTablePool;

    @Before
    public void setup() throws IOException {
       int poolSize=10;
        PoolMap.PoolType poolType = PoolMap.PoolType.ThreadLocal; // for example
        Configuration conf = new Configuration();
        // <可选参数>
        conf.set("rs.list.acquire.read.timeout", "10000"); // 同上
        conf.set("rs.list.acquire.connect.timeout", "10000"); 
        conf.set("rpc.connect.timeout", "1000"); 
        conf.set("rpc.execute.timeout", "1000");
        ohTablePool = new OHTablePool(conf, poolSize, PoolType);
        // <必须参数方式>
        ohTablePool.setParamUrl("tablename", "your ParamUrl");
        ohTablePool.setFullUserName("tablename", "your FullUserName");
        ohTablePool.setPassword("tablename", "your Password");
        ohTablePool.setSysUserName("tablename", "your setSysUserName";
        ohTablePool.setSysPassword("tablename", "your SysPassword";
    
        hTable = ohTablePool.getTable("tablename");
    }
}

支持的 HBase 操作

本章节主要介绍了 OHTable 接口操作叙述，如果想了解更多的接口信息，请参考：OHTable.java。相关 HBaseAPI 使用的 Demo 请参考：HBaseAPI 使用 Demo。

Refresh Table Entry

函数说明：重新刷新客户端路由表。函数原型：

void refreshTableEntry(String familyString, boolean hasTestLoad)
void refreshTableEntry(final String tableName, final String family, boolean hasTestLoad)

参数列表：

tableName：目标表名，在使用 ohTablePool 时需要指定。
familyString：目标列簇。
hasTestLoad：是否有压测表，由客户端 HBASE_HTABLE_TEST_LOAD_ENABLE 配置项决定，如不配置，该选项设置为 false。

操作示例：

// 使用 OHTable 进行初始化
 ((OHTable) hTable).refreshTableEntry("family", false);
 ((OHTable) hTable).refreshTableEntry("family", true);

// 使用 OHTableClient 进行初始化
((OHTableClient) hTable).refreshTableEntry("family", false);
((OHTableClient) hTable).refreshTableEntry("family", true);

// 使用 OHTablePool 进行初始化
ohTablePool.refreshTableEntry("test", "family", false);
ohTablePool.refreshTableEntry("test", "family", true);

Increment Column Value

函数说明：对表中的某行单列数据进行自增，执行成功则返回自增后的值。函数原型：

long incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount, boolean writeToWAL)

参数列表：

row：行键
family：目标列簇，这里是建表时指定的列簇
qualifier：目标列名
amount：增加的数量（可以为负值）
writeToWAL：是否预写日志（默认写入）

操作示例：

// 这里展示的是对单列自增 1
String column = "incrementColumn";
String key = "incrementKey";
String family = "family";
int increment_value = 1;
long ret = hTable.incrementColumnValue(
    key.getBytes(), 
    family.getBytes(),
    column.getBytes(), 
    increment_value,
    1);

Increment

函数说明：对指定行中的单列或多列进行增加（针对数值类型，如 long，int 等）。函数原型：

Result increment(Increment increment)

参数列表：

increment： Increment 类型，需要使用 addColumn 进行属性设置。

操作示例：

// 这里展示的是对单列自增 1
String column = "incrementColumn";
String key = "incrementKey";
String family = "family";
int increment_value = 1;
Increment increment = new Increment(key.getBytes());
increment.addColumn(family.getBytes(), column.getBytes(), increment_value);
Result r = hTable.increment(increment);

Append

函数说明：对指定行中的单列或多列进行增加（针对字符类型，例如 byte，string 等）。函数原型：

Result append(Append append)

参数列表：

append：Append 类型，需要使用 add 函数进行属性设置。

操作示例：

String column = "appendColumn";
String key = "appendKey";
String family = "family";
Append append = new Append(key.getBytes());
append.add(family.getBytes(), column.getBytes(), toBytes("_append"));
Result r = hTable.append(append);
Assert.assertEquals(1, r.raw().length);
for (KeyValue kv : r.raw()) {
    Assert.assertEquals("_append", Bytes.toString(kv.getValue()));
}

Check And Delete

函数说明：匹配删除列，若匹配条件符合并删除成功，返回 true，反之返回 false。函数原型：

boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier, byte[] value, Delete delete)

参数列表：

row：目标匹配行键
family：目标列簇，这里是建表时指定的列簇
qualifier：目标匹配列名
value：目标匹配列值
delete：预期需要删除的列

操作示例：

String key = "deleteKey";
String family = "family";
String column = "column";
delete = new Delete(key.getBytes());
delete.deleteColumn(family.getBytes(), column.getBytes());
boolean ret = hTable.checkAndDelete(
    key.getBytes(), 
    family.getBytes(), 
    column.getBytes(),
    value.getBytes(), 
    delete);

Delete

函数说明：删除指定的存储单元（cells）或行（rows）。函数原型：

void delete(Delete delete)
void delete(List<Delete> deletes)

参数列表：

deletes：Delete 对象，这里是指需要执行删除的行，通过 deleteFamily 函数设置。

操作示例：

// 如下是删除某 key，family，column 对应的行
String key = "delete_row";
String family = "family";
String column = "column";
Delete delete = new Delete(key.getBytes());
delete.deleteFamily(family.getBytes(), column.getBytes());
hTable.delete(delete);

Check And Put

函数说明：检查并替换指定列数据，如果条件满足并替换成功，返回 true，反之返回 false。函数原型：

boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier, byte[] value, Put put)

参数列表：

row：目标行键
family：目标列簇
qualifier：目标列名
value：目标列值
put：需要替换的列值

操作示例：

String key = "putkey";
String family = "family"
String column = "column";
String value = "value";
String new_value = "value_new"
put = new Put(key.getBytes());
put.add(family.getBytes(), column.getBytes(), new_value.getBytes());
boolean ret = hTable.checkAndPut(
    key.getBytes(), 
    family.getBytes(), 
    column.getBytes(),
    value.getBytes(), 
    put);

Put

函数说明：插入数据。函数原型：

void put(Put put)
void put(List<Put> puts)

参数列表：

put：Put 对象，用于插入单条记录。
puts：Put 对象集合，用于批量插入多条记录。

操作示例：

// 插入单条记录
String key = "putKey";
String family = "family";
String column = "column";
String value = "putValue";
Put put = new Put(key.getBytes());
put.add(family.getBytes(), column.getBytes(), value.getBytes());
hTable.put(put);

// 插入多条记录
String column = "column";
String value = "value";
String family = "family";
Put put1 = new Put(Bytes.toBytes("testKey1"));
put1.add(toBytes(family), toBytes(column), toBytes(value));
put1.add(toBytes(family), toBytes(column), System.currentTimeMillis(), toBytes(value));

Put put2 = new Put(Bytes.toBytes("testKey2"));
put2.add(toBytes(family), toBytes(column), toBytes(value));
put2.add(toBytes(family), toBytes(column), System.currentTimeMillis(), toBytes(value));
List<Put> puts = new ArrayList<Put>();
puts.add(put1);
puts.add(put2);
hTable.put(puts);

Scan

函数说明：根据指定条件（scan/family/qualifier）对 table 进行扫描。函数原型：

ResultScanner getScanner(byte[] family, byte[] qualifier)
ResultScanner getScanner(final byte[] family)
ResultScanner getScanner(final Scan scan)

参数列表：

family：目标列簇，这里是指需要过滤的列簇。
qualifier：目标列名，这里指需要过滤的列名。
scan：Scan 对象，使用 addFamily 或 addColumn 等设置操作目标。

操作示例：

// 指定scan
String startKey = "startKey";
String endKey = "endKey";
String family = "family";
String column = "column";
int maxVersion = 1;
Scan scan = new Scan();
scan.addColumn(family.getBytes(), column.getBytes());
scan.setMaxVersions(maxVersion);
scan.setStartRow(startKey.getBytes());
scan.setStopRow(endKey.getBytes());
ResultScanner scanner = hTable.getScanner(scan);
for (Result result : scanner) {
    for (KeyValue keyValue : result.raw()) {
        // User operation, for example: 
        System.out.println("rowKey: " + new String(keyValue.getRow())
            + " columnQualifier:" + new String(keyValue.getQualifier())
            + " timestamp:" + keyValue.getTimestamp() + " value:"
            + new String(keyValue.getValue()));
    }
}

Get

函数说明：获取指定行的数据。函数原型：

Result get(final Get get)
Result[] get(List<Get> gets)

参数列表：

get：Get 对象，需要使用 addColumn 或 addFamily 来指定操作目标。
gets：Get 对象列表，保存 Get 对象。

操作示例：

// 获取 one family
String key = "getKey";
String family = "family";
int maxVersion = 1;
Get get = new Get(key.getBytes());
get.addFamily(family.getBytes());
get.setMaxVersions(maxVersion);
Result result = hTable.get(get);

Exists

函数说明：判断在 Get 对象中设置的列簇或列是否存在，存在则返回 true，反之返回 false。函数原型：

boolean exists(Get get)

参数列表：

get：Get 对象，使用 addFamily 或 addColumn 来设置操作目标。

操作示例：

String key = "existKey";
String family = "family";
get = new Get(key.getBytes());
get.addFamily(family.getBytes());
Assert.assertTrue(hTable.exists(get));

Get Configuration

函数说明：返回操作实例中的 config 具柄，可用于在操作中变更属性。函数原型：

Configuration getConfiguration()

参数列表：

无

操作示例：

hTable.getConfiguration().set(HBASE_HTABLE_TEST_LOAD_ENABLE, "true");
hTable.getConfiguration().set(HBASE_HTABLE_TEST_LOAD_SUFFIX, "_a");
hTable.getConfiguration().set("rs.list.acquire.read.timeout", "10000");