ネイティブFlinkによるOBKV-HBaseへのアクセス

/*
* 現在のクラスタを次のように仮定します。
* ClusterName：obkvcluster
* TenantName：obkv
* DataBaseName: test
* UserName：root
* UserPassWord:
* SYS_USER_NAME : sysroot
* SYS_USER_PASSWD:
*/
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.client.connection.impl", "com.alipay.oceanbase.hbase.util.OHConnectionImpl")
conf.set("hbase.oceanbase.fullUserName", "root@obkv#obkvcluster");
conf.set("hbase.oceanbase.password", "");
// obconfig serverからRSlistのURLを取得する。詳細は後述のConfig URLの取得に関する内容を参照
conf.set("hbase.oceanbase.paramURL", "");
conf.set("hbase.oceanbase.sysUserName", "sysroot");
conf.set("hbase.oceanbase.sysPassword", "");

/*
* 現在のクラスタは以下のように仮定します。
* DataBaseName: test
* UserName：root
*/
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.client.connection.impl", "com.alipay.oceanbase.hbase.util.OHConnectionImpl")
conf.set(HBASE_OCEANBASE_FULL_USER_NAME, "root");
// ユーザーパスワード
conf.set(HBASE_OCEANBASE_PASSWORD, "");
// ODPアドレスの詳細は備考を参照
conf.set(HBASE_OCEANBASE_ODP_ADDR, "");
// OBKVのポートは3307です（固定）
conf.setInt(HBASE_OCEANBASE_ODP_PORT, "3307");
// クラウドではODPモードを使用します（固定）
conf.setBoolean(HBASE_OCEANBASE_ODP_MODE, true);
// データベースの名前
conf.set(HBASE_OCEANBASE_DATABASE, "test");

<!-- OBKV-HBase 依存関係には、hbase-client と hadoop-common が含まれています -->
<dependency>
  <groupId>com.oceanbase</groupId>
  <artifactId>obkv-hbase-client</artifactId>
  <version>2.2.1</version>
</dependency>

<!-- Flink Connector HBase 依存関係 -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-hbase-2.2</artifactId>
  <version>{flink.connector.hbase.version}</version>
  <exclusions>
    <exclusion>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-client</artifactId>
    </exclusion>
  </exclusions>
</dependency>

-- テーブルグループを作成
CREATE TABLEGROUP `flink_test` SHARDING = 'ADAPTIVE';

-- テーブルを作成
CREATE TABLE `flink_test$cf1` (
    `K` varbinary(1024) NOT NULL,
    `Q` varbinary(256) NOT NULL,
    `T` bigint(20) NOT NULL,
    `V` varbinary(1024) DEFAULT NULL,
    PRIMARY KEY (`K`, `Q`, `T`)
) TABLEGROUP = `flink_test` partition by key(`K`) partitions 17;
CREATE TABLE `flink_test$cf2` (
    `K` varbinary(1024) NOT NULL,
    `Q` varbinary(256) NOT NULL,
    `T` bigint(20) NOT NULL,
    `V` varbinary(1024) DEFAULT NULL,
    PRIMARY KEY (`K`, `Q`, `T`)
) TABLEGROUP = `flink_test` partition by key(`K`) partitions 17;

StreamExecutionEnvironment env =
        StreamExecutionEnvironment.getExecutionEnvironment();
env.fromElements("row1", "row2")
   .flatMap(new RichFlatMapFunction<String, String>() {
    private	transient Table table;
    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        Connection connection = ConnectionFactory.createConnection(ObHTableTestUtil.newConfiguration());
        table = connection.getTable(org.apache.hadoop.hbase.TableName.valueOf("flink_test"));
    }
    @Override
    public void flatMap(String s, Collector<String> out) throws Exception {
        // Write to HBase
        Put put = new Put(Bytes.toBytes(s));
        put.addColumn(Bytes.toBytes(family1), Bytes.toBytes(column1), Bytes.toBytes("cf1_" + s +"_value"));
        put.addColumn(Bytes.toBytes(family2), Bytes.toBytes(column2), Bytes.toBytes("cf2_" + s + "_value"));
        table.put(put);
        // Read from HBase
        Get get = new Get(Bytes.toBytes(s));
        Result result = table.get(get);
        String family1Value = Bytes.toString(result.getValue(Bytes.toBytes(family1), Bytes.toBytes(column1)));
        String family2Value = Bytes.toString(result.getValue(Bytes.toBytes(family2), Bytes.toBytes(column2)));
        // collect get result
        out.collect(String.format("Get result from cf1: Row: %s, Qualifier: %s, Value: %s", Bytes.toString(result.getRow()), column1, family1Value));
        out.collect(String.format("Get result from cf2: Row: %s, Qualifier: %s, Value: %s", Bytes.toString(result.getRow()), column2, family2Value));
    }
    @Override
    public void close() throws Exception {
        table.close();
    }
}).print();

CREATE TABLEGROUP `student_info` SHARDING = 'ADAPTIVE';
CREATE TABLEGROUP `student_hobby` SHARDING = 'ADAPTIVE';
CREATE TABLEGROUP `all_student_info` SHARDING = 'ADAPTIVE';
CREATE TABLE `student_info$cf1` (
    `K` varbinary(1024) NOT NULL,
    `Q` varbinary(256) NOT NULL,
    `T` bigint(20) NOT NULL,
    `V` varbinary(1024) DEFAULT NULL,
    PRIMARY KEY (`K`, `Q`, `T`)
) TABLEGROUP = `student_info` partition by key(`K`) partitions 17;
CREATE TABLE `student_hobby$cf1` (
    `K` varbinary(1024) NOT NULL,
    `Q` varbinary(256) NOT NULL,
    `T` bigint(20) NOT NULL,
    `V` varbinary(1024) DEFAULT NULL,
    PRIMARY KEY (`K`, `Q`, `T`)
) TABLEGROUP = `student_hobby` partition by key(`K`) partitions 17;
CREATE TABLE `all_student_info$cf1` (
    `K` varbinary(1024) NOT NULL,
    `Q` varbinary(256) NOT NULL,
    `T` bigint(20) NOT NULL,
    `V` varbinary(1024) DEFAULT NULL,
    PRIMARY KEY (`K`, `Q`, `T`)
) TABLEGROUP = `all_student_info` partition by key(`K`) partitions 17;

-- FlinkとHBaseのテーブルstudent_infoのマッピングを作成
CREATE TABLE student_info (
  id INT,
  cf1 ROW<name STRING, age INT>,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'hbase-2.2',
  'table-name' = 'student_info',
  'zookeeper.quorum' = 'any_string'
);

-- FlinkとHBaseのテーブルstudent_hobbyのマッピングを作成
CREATE TABLE student_hobby (
  id INT,
  cf1 ROW<hobby STRING>,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'hbase-2.2',
  'table-name' = 'student_hobby',
  'zookeeper.quorum' = 'any_string'
);

-- student_infoとstudent_hobbyにデータを挿入
INSERT INTO student_info VALUES
  (1001, ROW('Lily', 13)),
  (1002, ROW('John', 12)),
  (1003, ROW('David', 13));
INSERT INTO student_hobby VALUES
  (1001, ROW('basketball')),
  (1002, ROW('badminton')),
  (1003, ROW('jogging'));

-- FlinkとHBaseのテーブルall_student_infoのマッピングを作成
CREATE TABLE all_student_info (
  id INT,
  cf1 ROW<name STRING, age INT, hobby STRING>,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'hbase-2.2',
  'table-name' = 'all_student_info',
  'zookeeper.quorum' = 'any_string'
);

-- all_student_infoにデータを挿入
INSERT INTO all_student_info
  SELECT
    si.id AS id,
    ROW(si.cf1.name, si.cf1.age, sh.cf1.hobby) AS cf1
  FROM
    student_info AS si
  LEFT JOIN
    student_hobby AS sh ON si.id = sh.id;