兼容 HBase 的接口 - OceanBase 数据库 V4.3.5

OBKV-HBase 客户端实现了 HTableInterface 的接口。本文介绍了 OBKV-HBase 对开源 HBase 接口的兼容性。

接口兼容性

修饰符和类型	方法和描述	兼容性
Result	append (Append append）在单个行中追加一个或多个列的值。	支持
Object[]	batch(List<? extends Row> actions) 与 batch(List, Object[])相同，但返回结果数组，而不是使用结果参数引用。	不支持
void	batch(List<? extends Row> actions, Object[] results) 批量调用 Deletes、Gets、Puts、Increments、Appends 和 RowMutations 的方法。	不支持
boolean	checkAndDelete (byte[] row, byte[] family, byte[] qualifier, byte[] value,Delete delete) 原子性地检查行/列族/限定符值是否与预期值匹配。	支持
boolean	checkAndPut (byte[] row, byte[] family, byte[] qualifier, byte[] value, Put put) 原子性地检查行/列族/限定符值是否与预期值匹配。	支持
void	close() 释放内部缓冲区中的任何资源或待定的更改。	支持
<T extends CoprocessorProtocol,R>Map<byte[],R>	coprocessorExec(Class protocol, byte[] startKey, byte[] endKey, Batch.Call<T,R> callable) 调用传入的 Batch.Call，针对在选定区域中运行的 CoprocessorProtocol 实例。	不支持
<T extends CoprocessorProtocol,R> void	coprocessorExec(Class protocol, byte[] startKey, byte[] endKey, Batch.Call<T,R> callable, Batch.Callback callback) 针对在选定区域中运行的 CoprocessorProtocol 实例，调用传递的 Batch.Call。	不支持
<T extends CoprocessorProtocol > T	coprocessorProxy(Class protocol, byte[] row) 创建并返回在包含指定行的区域中运行的 CoprocessorProtocol 实例的代理。	不支持
void	delete(Delete delete) 删除指定的单元格/行。	支持
void	delete(List<Delete> deletes) 批量删除指定的单元格/行。	支持
boolean	exists(Get get) 检查表中是否存在Get指定的列。	支持
void	flushCommits() 执行所有缓冲的Put操作。	支持
Result	get(Get get) 从给定的行中提取特定的单元格。	支持
Result[]	get(List<Get> gets) 从给定的行中批量提取特定的单元格。	支持
org.apache.hadoop.conf.Configuration	getConfiguration() 返回此实例使用的 Configuration 对象。	支持
Result	getRowOrBefore(byte[] row, byte[] family) 已弃用。自版本 0.92 起，此方法已弃用，没有替换。getRowOrBefore 用于在 .META. 中查找条目，并做出各种假设（这些假设对于 .META. 是正确的，但不适用于一般情况）以提高效率。	Deprecated
ResultScanner	getScanner(byte[] family) 获取给定列族的扫描器。	支持
ResultScanner	getScanner(byte[] family, byte[] qualifier) 获取给定列族和限定符的扫描器。	支持
ResultScanner	getScanner(Scan scan) 根据Scan对象的指定，返回当前表上的一个扫描器。	支持
HTableDescriptor	getTableDescriptor() 获取此表的 table descriptor。	不支持
byte[]	getTableName() 获取此表的名称。	支持
long	getWriteBufferSize() 返回此 HTable 的最大写缓冲区大小（以字节为单位）。	支持
Result	increment(Increment increment) 在单个行中增加一个或多个列的值。	支持
long	incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount) 原子性地增加一个列的值。	支持
long	incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount, Durability durability) 原子性地增加一个列的值。	支持
boolean	isAutoFlush() 检查是否已启用'自动刷新'。	支持
RowLock	lockRow(byte[] row) 获取行锁。	不支持
void	mutateRow(RowMutations rm) 对单行执行多个原子操作。	不支持
void	put(List<Put> puts) 批量插入数据。	支持
void	put(Put put) 插入数据。	支持
void	setAutoFlush(boolean autoFlush) 设置是否自动刷新。	支持
void	setAutoFlush(boolean autoFlush, boolean clearBufferOnFail) 设置是否自动刷新，以及在失败时是否清空缓冲区。	支持
void	setWriteBufferSize(long writeBufferSize) 设置写缓冲区大小。	支持
void	unlockRow(RowLock rl) 释放行锁。	不支持
void	setBatch，设置批处理	支持
void	setCaching，设置缓存	支持
void	setMaxResultSize，设置最大结果大小	支持
void	setAllowPartialResults，设置是否允许部分结果	支持
void	checkExistenceOnly，设置是否只检查存在性	支持
void	checkAndMutate ，设置是否检查并修改	支持
void	connection ，设置连接	支持
void	ohBufferedMutatorImpl，设置缓冲修改器实现	支持

接口使用差异

因为架构实现上的差异，开源 HBase 的代码迁移到 OBKV-HBase，需要调整一些配置代码。

差异说明

开源 HBase 和 OBKV-HBase 数据访问代码没有变化，数据访问部分兼容。不同点主要在：

Configuration：开源 HBase 和 OBKV-HBase 架构不同，配置项也不同。
OHTableClient：需要基于 OBKV-HBase 的句柄来做数据访问。

代码示例

开源 HBase 的代码：

public class HBaseInsertDemo {
    public static void main(String[] args) throws Exception {
        // 配置HBase
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "your_zookeeper_quorum"); // 设置ZooKeeper地址
        conf.set("hbase.zookeeper.property.clientPort", "2181"); // 设置ZooKeeper端口
        // 拿到Table句柄
        TableName tableName = TableName.valueOf("your_table_name"); 
        HTable table =new HTable(tableName, conf);
        // 准备插入数据
        Put put = new Put(Bytes.toBytes("row_key"));
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("qualifier"), Bytes.toBytes("value")); // 列族，列标识符，值
        table.put(put);
        // 关闭资源
        table.close();
        System.out.println("数据插入成功！");
    }
}

OBKV-HBase 的代码：

public class OBHBaseInsertDemo {
    public static void main(String[] args) throws Exception {
        // 配置HBase
        Configuration conf = new Configuration();
        conf.set(HBASE_OCEANBASE_FULL_USER_NAME, "");
        conf.set(HBASE_OCEANBASE_PASSWORD, "");
        conf.set(HBASE_OCEANBASE_PARAM_URL, "");
        conf.set(HBASE_OCEANBASE_SYS_USER_NAME, "");
        conf.set(HBASE_OCEANBASE_SYS_PASSWORD, "");
        // 拿到Table句柄
        TableName tableName = TableName.valueOf("your_table_name"); 
        OHTableClient table = new OHTableClient(tableName, conf); // 创建 htable对象

        // 准备插入数据
        Put put = new Put(Bytes.toBytes("row_key"));
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("qualifier"), Bytes.toBytes("value")); // 列族，列标识符，值
        table.put(put);
        // 关闭资源
        table.close();
        System.out.println("数据插入成功！");
    }
}

接口说明及示例

scan 类的接口说明及示例

支持 scan 类的一些接口，包括 setBatch/setCaching/setMaxResultSize/setAllowPartialResults/isCacheBlock

setBatch

用于设置每次调用 next() 返回的最大 cell 的个数，用于在客户端实现分页查询。

public scan setBatch(int batch);

setCaching

用于限制单次 RPC 服务端返回的行数，默认是没有限制。客户可以使用较大的 cahing 值增加每次 RPC 获取的数据量，来加快查询的速率，也可以使用较小的 cahing 值来减少客户端的内存消耗，避免 OOM。

public scan setCaching(int caching);

setMaxResultSize

用于限制单次 RPC 服务端返回数据的最大字节数，默认是没有限制。

public scan setMaxResultSize(long maxResultSize);

setAllowPartialResults

设置是否允许返回部分行给用户，避免单行数据过长导致客户端 OOM，默认是 false，这时即使用户设置了 setMaxResultSize 来限制返回的数据量，但是每次服务端还是会至少返回一整行数据给客户端。

public scan setAllowPartialResults(final boolean allowPartialResults);

setCacheBlocks

用于设置查询的结果集是否缓存到 OB 的 rowCache/BlockCache 中，默认为 true。比如再做全表扫描的时候可以设置为 false 来避免污染缓存。

public scan setCacheBlocks(boolean cacheBlocks);

renew

用于设置查询结果集的过期时间，默认是 0，表示不设置过期时间。

ResultScanner.renewLease() 方法的作用是延长 ResultScanner 的有效期限（lease）。在 HBase 中，ResultScanner 自带一个默认的有效期限（60000ms），如果在这个期限内没有读取数据，扫描器将自动关闭。通过调用 renewLease() 方法，可以通过刷新 ResultScanner 状态的方式重新计数有效期限，延长扫描器开启时间。

OBKV-HBase 中，ResultScanner 的有效期限通过kv_hbase_client_scanner_timeout_period调整。

示例如下：

ResultScanner scanner = hTable.getScanner(scan);

Thread.sleep(5 * 1000);
scanner.next();
Thread.sleep(20 * 1000);
scanner.renewLease();

Thread.sleep(41 * 1000);
scanner.next();

withStartRow/withStopRow

Scan.withStartRow 和 Scan.withStopRow 方法用于指定扫描的起始行和结束行，包括是否包含这些行键。这些方法可以帮助你限定扫描的范围，确保只扫描特定行键范围内的数据。

示例如下：

// 创建 Scan 对象
Scan scan = new Scan();

// 设置起始行和结束行
byte[] startRow = Bytes.toBytes("row1");
byte[] stopRow = Bytes.toBytes("row5");

scan.withStartRow(startRow);
scan.withStopRow(stopRow);

// 获取 ResultScanner
ResultScanner scanner = table.getScanner(scan);

setLimit

Scan.setLimit(int limit) 方法用于设置扫描的最大结果数。通过这个方法，可以限制每次扫描返回的结果数量，防止一次性返回大量数据导致内存溢出。

示例如下：

// 创建 Scan 对象
Scan scan = new Scan();

// 设置最大结果数
int limit = 100;
scan.setLimit(limit);
// 获取 ResultScanner
ResultScanner scanner = table.getScanner(scan);

get 类的接口说明及示例

checkExistenceOnly

用于设置执行 Get 返回的 Result 是否只包含 exists 标志（通过 Result 的 getExists 接口获取）而不是结果集，用于检查是否存在满足条件的数据。

public get setCheckExistenceOnly(boolean checkExistenceOnly);

checkAndMutate

checkAndMutate 是 table 的一个接口，可以进行同一个 row 的批量操作。指定的 checkAndMutate 操作可以通过一次 RPC 发送到 RegionServer，不同于原生的 HBase，如果其中一个操作失败，则其余操作全部都会回滚。

场景：当需要校验和插入是一个原子操作的时候，比如先校验该人的钱包余额是否大于 100，如果大于则修改该余额，返回 true，如果不大于，则不修改返回 false。

参数：

row：操作的行。
family、qualifier、value：需要校验的 cell。
compareOp：比较运算符。
rowMutations：如果该行符合，则进行的操作。

true 或 false，表示是否匹配成功。

connection

该特性是给用户提供一个连接数据库的句柄，方便用户从连接中获得各个 table 的句柄，从而操作数据。或者提供一些连接级别的接口，如 admin、bufferMutate 操作等。

Table 和 BufferMutator

下面展示一段 HBase 常用的连接用法代码：

Configuration conf = HBaseConfiguration.create();

conf.set(HConstants.ZOOKEEPER_QUORUM, "zk1, zk2, zk3");
conf.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, 2181);
conf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/hbase");

//Connection为线程安全对象，在整个程序的生命周期里，只需构造一个Connection对象
//在程序结束后，需要将Connection对象关闭，否则会造成连接泄露。可以采用try finally方式防止泄露
Connection connection = ConnectionFactory.createConnection(conf);
TableName tableName = TableName.valueOf("tablename");
// 获取Table句柄
Table table = connection.getTable(tableName);
// 获取BufferMutator
BufferedMutator ohBufferMutator = connection.getBufferedMutator(TableName);

HBase 是通过 ConnectionFactory 传入一个 config 创建一个 Connection 的，然后通过 Connection 取出对应的表进行操作。

方法一：直接设置 Configuration

conf.set(ClusterConnection.HBASE_CLIENT_CONNECTION_IMPL,
         "com.alipay.oceanbase.hbase.util.OHConnectionImpl");

方法二：修改配置文件

使用 hbase-site.xml、hbase-default.xml 里面增加这个配置项，优先级为 hbase-site.xml > hbase-default.xml。

<property>
  <name>hbase.client.connection.impl</name>
  <value>com.alipay.oceanbase.hbase.util.OHConnectionImpl</value>
</property>

TableBuilder

TableBuilder 是 HBase 2.x 用于定义和获取 HTable 的新结构。用户可以通过给 TableBuilder 设置 TableName，ConnectionConfiguration,rpcTimeout 等获取对应的 HTable。

示例如下：

获取 tableBuilder
```
public TableBuilder getTableBuilder(TableName tableName, ExecutorService pool);
```
注意：如果用户在调用 getTableBuilder 之前已经定义了 pool，则需要手动关闭 pool。HTable 的关闭不会自动关闭用户定义的 pool。

通过 tableBuilder 获取 OHTable

Configuration conf = ObHTableTestUtil.newConfiguration();
conf.set(...);
connection = ConnectionFactory.createConnection(conf);
ExecutorService pool = ...;
TableName tableName = TableName.valueOf("table_name");
TableBuilder builder = connection.getTableBuilder(tableName, pool);

设置连接超时时间

tabelClient 中的 rpcConnectTimeout 和 ObTable 中的 obTableConnectTimeout 参数控制着 Table 客户端在初始化时和当前租户所在的 observer 建立连接时的连接超时时间。tableClient 的连接超时时间最小为 1 秒。

OBKV-HBase 客户端设计：

OBKV-HBase 的 Table 客户端的两个参数都可以通过 ObTableClient 中的设置接口进行设置，所以需要在 HBase 客户端对 TableClient 对象 init 前进行设置。

/*
 * Set rpc connect timeout.
 */
public void setRpcConnectTimeout(int rpcConnectTimeout) {
    this.properties.put(RPC_CONNECT_TIMEOUT.getKey(), String.valueOf(rpcConnectTimeout));
    this.rpcConnectTimeout = rpcConnectTimeout;
}

为了兼容早期版本的参数 ipc.socket.timeout，用户依然可以通过该参数对连接超时时间进行设置，但新参数 hbase.ipc.client.socket.timeout.connect 的优先级会高于老参数。也就是说，只有在用户只设置了老参数而没有设置新参数的情况下，Table 客户端会使用老参数的设置时间。其他情况都会使用新参数的默认值或设置值。

// SOCKET_TIMEOUT_CONNECT = "hbase. ipc. client. socket. timeout. connect"，默认值为 10s
// SOCKET_TIMEOUT = "ipc. socket. timeout"，默认值为 20s 
int rpcConnectTimeout = -1;
// 如果设置了新参数，直接使用
if (conf.get(SOCKET_TIMEOUT_CONNECT) != null) {
    rpcConnectTimeout = conf
            .getInt(SOCKET_TIMEOUT_CONNECT, DEFAULT_SOCKET_TIMEOUT_CONNECT);
} else {
    // 如果没有设置新参数，但是设置了老参数，用老参数
    if (conf.get(SOCKET_TIMEOUT) != null) {
        rpcConnectTimeout = conf.getInt(SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT);
    } else { // 两个都没设置，用新参数的默认值
        rpcConnectTimeout = conf
                .getInt(SOCKET_TIMEOUT_CONNECT, DEFAULT_SOCKET_TIMEOUT_CONNECT);
    }
}
this.rpcConnectTimeout = rpcConnectTimeout;

将 RpcConnectTimeout 传入创建 ObTableClient 的函数，在 init 前设置。

public static ObTableClient getOrCreateObTableClient(ObTableClientKey obTableClientKey,
                                                         int rpcConnectTimeout) throws IOException {
    // other code...
                obTableClient.setFullUserName(obTableClientKey.getFullUserName());
                obTableClient.setPassword(obTableClientKey.getPassword());
                obTableClient.setRpcConnectTimeout(rpcConnectTimeout);
                obTableClient.init();
                OB_TABLE_CLIENT_INSTANCE.put(obTableClientKey, obTableClient);
            }
        } catch (Exception e) {
            throw new IOException(e);
        } finally {
            lock.unlock();
        }
    }
    return OB_TABLE_CLIENT_INSTANCE.get(obTableClientKey);
}

OBKV-HBase 客户端会保存已经创建的 ObTableClient 对象，不会重新初始化，所以如果用户后续再想设置连接超时时间给 ObTableClient 不能用 Configuration.set 的方式，而需要直接使用对应的 ObTableClient 对象的 setRpcConnectTimeout 接口。

不兼容接口

当前不兼容接口的一些说明

getRegionServer 等均不兼容
目前 connection 只提供 getTable、getBufferdMutator、TableBuilder 三个接口

table 类的接口说明及示例

setRpcTimeout、getRpcTimeout:一次 rpc 调用的超时时间
setOperationTimeout、getOperationTimeout:一次操作的超时时间，可能包括重试或者其他带来的多次 rpc 调用

BufferedMutator 接口说明及示例

与 HBase 的 BufferedMutator 类似，都是用于累积 mutation 操作，然后批量执行。用户可以选择手动提交批处理操作，或者当 OBKV BufferedMutator 达到累积阈值时自动提交。ohBufferedMutatorImpl 是 OBKV-HBase 中的 BufferedMutator 实现类。

目前 OBKV-HBase 的 BufferedMutator 只支持删除和添加操作，且一个已创建的 OBKV-HBase 的 BufferedMutator 对象只能对一种 mutation 进行批量操作。BufferedMutator 中的操作无法保证并发顺序，在存在失败操作时也无法保证执行顺序和原子性。

1.x 兼容

OBKV-HBase 的 BufferedMutator 与 HBase 的 BufferedMutator 1.x 版本兼容。OHBufferedMutatorImpl 1.x 依赖 OHTable 1.x 中的 batch 接口。

功能说明如下：

同一个 BufferedMutator 对象能操作混合的 Put 和 Delete 操作。
BufferedMutator 中的 put 和 delete 操作均支持多列族。
当 BufferedMutator 中的 Mutation 操作数量达到阈值时，它会执行批处理操作。自动批处理操作有一个执行上限，这个上限与原生 HBase 一致，等于批处理阈值的两倍。
对于失败的 Mutation 操作，系统会进行回收处理。

2.x 兼容

OBKV-HBase 的 BufferedMutator 与 HBase 的 BufferedMutator 2.x 版本兼容。OHBufferedMutatorImpl 2.x 依赖 OHTable 2.x 中的 batch 接口。

功能说明如下：

同一个 BufferedMutator 对象能操作混合的 put 和 delete 操作。
BufferedMutator 中的 put 和 delete 操作均支持多列族。
BufferedMutator 内部有一个定时器，用于定期执行批处理操作。每次批处理操作都会提交所有已累积的操作。
对于失败的 Mutation 操作，系统会进行回收处理。

2.x 新增接口：

新增了 setWriteBufferPeriodicFlush 接口，用于设置自动提交器。timeoutMs 用于控制两次执行的间隙，timerTickMs 用于控制提交器触发间隙。示例如下：
```
public synchronized void setWriteBufferPeriodicFlush(long timeoutMs, long timerTickMs);
```
新增了 disableWriteBufferPeriodicFlush 接口，用于关闭自动提交器。示例如下：
```
public void disableWriteBufferPeriodicFlush();
```

获取 BufferedMutator 的方式

获取 BufferedMutator 的方式有两种：

传入 TableName，指定表名，其他采用默认参数

public BufferedMutator getBufferedMutator(TableName tableName) throws IOException {
    return getBufferedMutator(new BufferedMutatorParams(tableName));
}

传入BufferedMutatorParams，采用自定义参数

public BufferedMutator getBufferedMutator(BufferedMutatorParams params) throws IOException {
        if (params.getTableName() == null) {
            throw new IllegalArgumentException("TableName cannot be null.");
        }
        if (params.getPool() == null) {
            params.pool(HTable.getDefaultExecutor(getConfiguration()));
        }
        if (params.getWriteBufferSize() == BUFFERED_PARAM_UNSET) {
            params.writeBufferSize(connectionConfig.getWriteBufferSize());
        }
        if (params.getMaxKeyValueSize() == BUFFERED_PARAM_UNSET) {
            params.maxKeyValueSize(connectionConfig.getMaxKeyValueSize());
        }
        return new OHBufferedMutatorImpl(this, params);
    }

添加 mutation 操作：

单个 mutation 操作

public void mutate(Mutation mutation) throws IOException {
    mutate(Collections.singletonList(mutation));
}

多个 mutation 操作接口

public void mutate(List<? extends Mutation> mutations) throws IOException {
    // ...
}

在调用 close 后也可以手动 flush 提交已有操作，但是无法添加新 mutation 操作。

public void flush() throws IOException {
    asyncExecute(true);
}

CheckAndMutateBuilder 接口说明及示例

CheckAndMutateBuilder 是 HBase 2.x HTable 中的一种新结构，用于执行 CheckAndxxx 操作。用户可以通过设置行 row，列族 family，列 qualifier，value，时间范围 timeRange，比较符 CompareOp 等参数来执行 put，delete 和 mutate 操作。CheckAndMutateBuilder 会根据用户设置的 CompareOp 和 family 等参数与库中的数据进行比较，如果匹配成功，则执行相应的操作。

接口说明如下：

形式	作用
CheckAndMutateBuilder qualifier(byte[] qualifier)	设置 qualifier
CheckAndMutateBuilder timeRange(TimeRange timeRange)	设置 TimeRange
CheckAndMutateBuilder ifNotExists();	用于检查是否 qualifier 不存在
CheckAndMutateBuilder ifEquals(byte[] value)	用于检查是否存在 value
CheckAndMutateBuilder ifMatches(CompareOperator compareOp, byte[] value)	设置比较符和 value
boolean thenPut(Put put)	执行 put 操作
boolean thenDelete(Delete delete)	执行 delete 操作
boolean thenMutate(RowMutations mutation)	执行 put 和 delete 混合操作

使用示例如下：

key 为 putKey，family 为 family_group，qualifier 为 column1，value 为 value1 的数据时，进行 put 操作：

OHTable hTable;
Put put = new Put(toBytes("putKey"));
put.addColumn(toBytes("family_group"), toBytes("column1"), toBytes("value2"));
boolean ret = hTable.checkAndMutate(toBytes("putKey"), toBytes("family_group"))
                    .qualifier(toBytes("column1")
                    .ifEqual("value1")
                    .thenPut(put);

batch 接口说明及示例

batch(List<Row> actions, Object[] results) 方法用于批量执行多个操作（如 put、get、delete 等），并将结果存储在指定的数组中。通过这个方法，可以一次执行多个操作，提高效率。

当前只支持 put 和 delete 的操作，可以支持 put 和 delete 混合操作，以及多列族的操作。

接口说明

// 1.x 和 2.x 均有该接口
public void batch(final List<? extends Row> actions, final Object[] results)
// 2.x 移除了该接口
public void batch(final List<? extends Row> actions)

使用示例

List<Row> batchLsit = new LinkedList<>();
for (int i = 0; i < rows; ++i) {
    Put put = new Put(toBytes("Key" + i));
    Delete delete = new Delete(toBytes("Key" + i));
    batchLsit.add(delete);
    put.addColumn(family1, family1_column1, family1_value);
    put.addColumn(family1, family1_column2, family1_value);
    put.addColumn(family1, family1_column3, family1_value);
    put.addColumn(family2, family2_column1, family2_value);
    put.addColumn(family2, family2_column2, family2_value);
    put.addColumn(family3, family3_column1, family3_value);
    batchLsit.add(put);
}

// f1c1 f1c2 f1c3 f2c1 f2c2 f3c1
Delete delete = new Delete(toBytes("Key1"));
delete.addColumns(family1, family1_column1);
delete.addColumns(family2, family2_column1);
batchLsit.add(delete);
Object[] results = new Object[batchLsit.size()];
multiCfHTable.batch(batchLsit, results);

batchCallback 接口说明及示例

Table.batch(List<Row> actions, Batch.Callback<R> callback) 方法可以批量执行多个操作，并在每个操作完成时执行回调函数。这样，结果可以在每个操作完成后立即处理，而不是等到所有操作完成后再处理。

接口说明

// 1.x 和 2.x 均有该接口
public <R> void batchCallback(final List<? extends Row> actions, 
                              final Object[] results, 
                              final Batch.Callback<R> callback)
// 2.x 移除了该接口
public <R> void batchCallback(final List<? extends Row> actions, 
                              final Batch.Callback<R> callback)

使用示例

List<Row> batchLsit = new LinkedList<>();
for (int i = 0; i < rows; ++i) {
    Put put = new Put(toBytes("Key" + i));
    Delete delete = new Delete(toBytes("Key" + i));
    batchLsit.add(delete);
    put.addColumn(family1, family1_column1, family1_value);
    put.addColumn(family1, family1_column2, family1_value);
    put.addColumn(family1, family1_column3, family1_value);
    put.addColumn(family2, family2_column1, family2_value);
    put.addColumn(family2, family2_column2, family2_value);
    put.addColumn(family3, family3_column1, family3_value);
    batchLsit.add(put);
}

// f1c1 f1c2 f1c3 f2c1 f2c2 f3c1
Delete delete = new Delete(toBytes("Key1"));
delete.addColumns(family1, family1_column1);
delete.addColumns(family2, family2_column1);
batchLsit.add(delete);
Object[] results = new Object[batchLsit.size()];

multiCfHTable.batchCallback(batchLsit, results, new Batch.Callback<Result>() {
    @Override
    public void update(byte[] region, byte[] row, Result result) {
        updateCounter[0]++;
    }
});

Admin 接口说明及示例

OBKV-HBase 的 Admin 接口与 HBase 的 Admin 部分接口兼容。

注意

OBKV-HBase 从 V4.3.5 BP3 开始支持 Admin 接口。

准备工作

OBKV-HBase 的 Admin 接口功能需要通过 _enable_kv_hbase_admin_ddl 这个租户级隐藏配置项启用，该配置项默认值为 false，即关闭此功能。启用命令如下：

ALTER SYSTEM SET _enable_kv_hbase_admin_ddl = true;

createTable

createTable 用于创建 HBase 表。

在 OBKV-HBase 的底层实现中，会创建一个指定名称的表组（Tablegroup），并在该表组下创建 TableDescriptor 参数指定的列族，以及每个列族对应的 TTL、MaxVersion 参数。每个列族还对应一张实体表，实体表表名格式为 表名$列族名。创建表组和实体表是两个必要的步骤，系统会确保这两个步骤同时成功或同时失败。

接口定义如下：

void createTable(TableDescriptor desc) throws IOException

示例创建了一张 test 表，表下有 2 个列族（cf1 和 cf2），其中 cf1 的 MaxVersion 是 2，数据存活时间是 86400 秒（1 天）；cf2 的 MaxVersion 是 1，数据存活时间是 86400 秒（1 天）。具体如下：

Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();

// 1. 构造 htable 描述和列族描述
HColumnDescriptor hcd1 = new HColumnDescriptor(Bytes.toBytes("cf1"));
hcd1.setMaxVersions(2);
hcd1.setTimeToLive(86400);

HColumnDescriptor hcd2 = new HColumnDescriptor(Bytes.toBytes("cf2"));
hcd2.setMaxVersions(1);
hcd2.setTimeToLive(86400);

// 2. 执行创建表操作
HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("test"));
htd.addFamily(hcd1);
htd.addFamily(hcd2);
admin.createTable(htd);

注意事项如下：

目前仅支持 void createTable(TableDescriptor desc) 接口。
目前列族仅支持设置 TTL 和 MaxVersion。
不支持 cell 级别的 TTL。

tableExists

tableExists 用于查看指定 HBase 表是否存在，如果存在返回 true，如果不存在返回 false。如果 HBase 表带 namespace，需通过 namespace:tableName 的形式指定 namespace。namespace 映射到 OceanBase 集群中，表示 database。

接口定义如下：

public boolean tableExists(TableName tableName) throws IOException;

示例通过 tableExists 接口检查 test 表是否存在。

Configuration conf = ObHTableTestUtil.newConfiguration();
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
if (admin.tableExists(TableName.valueOf("test"))) {
    ...
}

getTableDescriptor

getTableDescriptor 是用于获取 HBase 表描述符的方法，在 OBKV-HBase 中用于检索表的元数据信息。元数据信息包含表名，以及列族级别的配置 HColumnDescriptor，目前主要包括列族名称、MaxVersion、TTL、表状态（enable/disable）。

接口定义如下：

HTableDescriptor getTableDescriptor(TableName tableName) throws TableNotFoundException, IOException;

示例通过 getTableDescriptor 接口获取 test 表的描述符，并打印表的列族信息。

Configuration conf = ObHTableTestUtil.newConfiguration();
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
HTableDescriptor descriptor = admin.getTableDescriptor(TableName.valueOf("test"));
if (descriptor.hasFamily("cf1".getBytes())) {
    ColumnFamilyDescriptor family1Desc = descriptor.getColumnFamily("cf1".getBytes());
    int ttl = family1Desc.getTimeToLive();
    int maxVersions = family1Desc.getMaxVersions();
}

isTableDisabled

isTableDisabled 用于检查指定 HBase 表是否被禁用，如果被禁用返回 true，否则返回 false。一般用于 enableTable 或其它操作前。

接口定义如下：

public boolean isTableDisabled(TableName tableName);

示例通过 isTableDisabled 接口检查 test 表是否被禁用，然后使用 enableTable 接口启用 test 表。

Configuration conf = ObHTableTestUtil.newConfiguration();
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
TableName tableName = TableName.valueOf("test");
if (admin.isTableDisabled(tableName)) {
    admin.enableTable(tableName);
}

isTableEnabled

isTableEnabled 用于检查指定 HBase 表是否被启用，如果被启用返回 true，否则返回 false。一般用于 disableTable 或其它操作前。

接口定义如下：

public boolean isTableEnabled(TableName tableName);

示例通过 isTableEnabled 接口检查 test 表是否被启用，然后使用 disableTable 接口禁用 test 表。

Configuration conf = ObHTableTestUtil.newConfiguration();
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
TableName tableName = TableName.valueOf("test");
if (admin.isTableEnabled(tableName)) {
    admin.disableTable(tableName);
}

disableTable

disableTable 用于禁用指定 HBase 表，在调用此接口后 HBase 表将处于不可用状态。disableTable 接口一般用于 HBase 表维护，比如清空数据、修改表结构等操作前会用 disableTable 禁用 HBase 表。

接口定义如下：

public void disableTable(TableName tableName);

示例先通过接口 isTableEnabled 判断 test 表是否是 enabled 状态，然后使用 disableTable 接口禁用表，最后通过接口 isTableDisabled 检查表是否是 disabled 状态。

Configuration conf = ObHTableTestUtil.newConfiguration();
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
TableName tableName = TableName.valueOf("test");
if (admin.isTableEnabled(tableName)) {
    admin.disableTable(tableName);
    Assert.assertTrue(admin.isTableDisabled(tableName));
}

deleteTable

deleteTable 用于删除指定 HBase 表，在调用此接口后表将处于不可用状态。

接口定义如下：

void deleteTable(TableName tableName) throws IOException

示例先通过 disableTable 接口禁用表 test，然后通过 deleteTable 接口删除 test 表。

Configuration conf = ObHTableTestUtil.newConfiguration();
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
TableName tableName = TableName.valueOf("test");
try {
    admin.disableTable(tableName);
    admin.deleteTable(tableName);
}

注意事项如下：

如果 HBase 表带 namespace，需通过 namespace:tableName 的形式指定 namespace。如果 namespace 不存在，将报 NamespaceNotFoundException 异常。namespace 映射到 OceanBase 集群中，表示 database。
如果 HBase 表不存在，将报错 TableNotFoundException。
删除前，需先通过 disableTable 接口禁用表，否则将报错 TableNotDisabledException。