兼容 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("数据插入成功！");
    }
}

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("数据插入成功！");
    }
}

public scan setBatch(int batch);

public scan setCaching(int caching);

public scan setMaxResultSize(long maxResultSize);

public scan setAllowPartialResults(final boolean allowPartialResults);

public scan setCacheBlocks(boolean cacheBlocks);

ResultScanner scanner = hTable.getScanner(scan);

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

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

// 创建 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);

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

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

public get setCheckExistenceOnly(boolean checkExistenceOnly);

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);

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

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

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

// 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;

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);
}

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

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);
    }

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

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

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

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);

// 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);

// 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]++;
    }
});

ALTER SYSTEM SET _enable_kv_hbase_admin_ddl = true;

void createTable(TableDescriptor desc) throws IOException

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);

public boolean tableExists(TableName tableName) throws IOException;

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

HTableDescriptor getTableDescriptor(TableName tableName) throws TableNotFoundException, IOException;

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();
}

public boolean isTableDisabled(TableName tableName);

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);
}

public boolean isTableEnabled(TableName tableName);

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);
}

public void disableTable(TableName tableName);

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));
}

void deleteTable(TableName tableName) throws IOException

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);
}