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オブジェクトは1つだけ作成すれば済みます。
//プログラム終了時には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"、デフォルト値は10秒
// SOCKET_TIMEOUT = "ipc. socket. timeout"、デフォルト値は20秒
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);
}