Flinkコネクタを使用してOBKV-HBaseにデータを同期する|V4.4.2|OceanBaseデータベース|ドキュメント|OceanBaseマニュアル

OBKV-HBaseのFlinkコネクタを使用すると、Flink内でobkv-hbase-client-javaを通じてOceanBaseにデータを書き込むことができます。

はじめに

公式リリース版はReleasesページまたはMaven中央リポジトリから入手できます。

<dependency>
    <groupId>com.oceanbase</groupId>
    <artifactId>flink-connector-obkv-hbase</artifactId>
    <version>${project.version}</version>
</dependency>

最新のスナップショットバージョンを使用したい場合は、Mavenスナップショットリポジトリを設定して指定できます。

<dependency>
    <groupId>com.oceanbase</groupId>
    <artifactId>flink-connector-obkv-hbase</artifactId>
    <version>${project.version}</version>
</dependency>

<repositories>
    <repository>
        <id>sonatype-snapshots</id>
        <name>Sonatype Snapshot Repository</name>
        <url>https://s01.oss.sonatype.org/content/repositories/snapshots/</url>
        <snapshots>
            <enabled>true</enabled>
        </snapshots>
    </repository>
</repositories>

ソースコードからビルドすることでもパッケージを取得できます。

git clone https://github.com/oceanbase/flink-connector-oceanbase.git
cd flink-connector-oceanbase
mvn clean package -DskipTests

SQL JAR

Flink SQLからこのコネクタを直接使用するには、すべての依存関係を含む flink-sql-connector-obkv-hbase-${project.version}.jar という名前のJARファイルをダウンロードする必要があります。

例

OceanBaseにテーブルhtable1$family1を作成します。このテーブル名では、htable1はHBaseのテーブル名に、family1はHBaseのcolumn familyに対応しています。

use test;
CREATE TABLE `htable1$family1`
(
  `K` varbinary(1024)    NOT NULL,
  `Q` varbinary(256)     NOT NULL,
  `T` bigint(20)         NOT NULL,
  `V` varbinary(1048576) NOT NULL,
  PRIMARY KEY (`K`, `Q`, `T`)
)

Javaアプリケーションの例

Mavenプロジェクトを例に、アプリケーションのpom.xmlファイルに必要な依存関係を追加し、次のコードを使用します。

package com.oceanbase;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Main {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tEnv =
                StreamTableEnvironment.create(
                        env, EnvironmentSettings.newInstance().inStreamingMode().build());

        tEnv.executeSql(
                "CREATE TABLE t_sink ( "
                        + " rowkey STRING,"
                        + " family1 ROW<column1 STRING, column2 STRING>,"
                        + " PRIMARY KEY (rowkey) NOT ENFORCED"
                        + ") with ("
                        + "  'connector'='obkv-hbase',"
                        + "  'url'='http://127.0.0.1:8080/services?...',"
                        + "  'schema-name'='test',"
                        + "  'table-name'='htable1',"
                        + "  'username'='root@test',"
                        + "  'password'='',"
                        + "  'sys.username'='root',"
                        + "  'sys.password'=''"
                        + ");");

        tEnv.executeSql(
                        "INSERT INTO t_sink VALUES "
                                + "('1', ROW('r1f1c1', 'r1f1c2')),"
                                + "('2', ROW('r2f1c1', 'r2f1c2')),"
                                + "('2', ROW('r3f1c1', 'r3f1c2'));")
                .await();
    }
}

実行が完了したら、OceanBaseで検索して検証できます。

詳細については、OBKVHBaseConnectorITCase.java を参照してください。

Flink SQLの例

使用する依存JARファイルをFlinkのlibディレクトリに配置し、その後SQL Clientを使用してFlink内に対象テーブルを作成します。

CREATE TABLE t_sink
(
  rowkey STRING,
  family1 ROW <column1 STRING,
  column2 STRING >,
  PRIMARY KEY (rowkey) NOT ENFORCED
) with (
  'connector'='obkv-hbase',
  'url'='http://127.0.0.1:8080/services?...',
  'schema-name'='test',
  'table-name'='htable1',
  'username'='root@test',
  'password'='',
  'sys.username'='root',
  'sys.password'='');

テストデータを挿入します。

INSERT INTO t_sink
VALUES ('1', ROW ('r1f1c1', 'r1f1c2')),
       ('2', ROW ('r2f1c1', 'r2f1c2')),
       ('2', ROW ('r3f1c1', 'r3f1c2'));

実行完了後、OceanBaseデータベースで検索して検証できます。

構成パラメータ

パラメータ名	必須	デフォルト値	タイプ	説明
url	はい		String	クラスタのconfig url。`SHOW PARAMETERS LIKE 'obconfig_url'`で確認できます。
schema-name	はい		String	OceanBaseのdb名。
table-name	はい		String	HBaseテーブル名。OceanBaseではテーブル名の構造は `hbase_table_name$family_name` です。
username	はい		String	非sysテナントのユーザー名。
password	はい		String	非sysテナントのパスワード。
sys.username	はい		String	sysテナントのユーザー名。
sys.password	はい		String	sysテナントユーザーのパスワード。
hbase.properties	いいえ		String	`'obkv-hbase-client-java'` のプロパティを設定します。複数の値はセミコロンで区切ります。
sync-write	いいえ	false	Boolean	同期書き込みを有効にするかどうか。`true` に設定すると、バッファを使用せずに直接データベースに書き込みます。
buffer-flush.interval	いいえ	1s	Duration	バッファフラッシュの間隔。`'0'` に設定すると、定期的なフラッシュを無効にします。
buffer-flush.buffer-size	いいえ	1000	Integer	バッファサイズ。
max-retries	いいえ	3	Integer	失敗時のリトライ回数。

はじめに

SQL JAR

例

Javaアプリケーションの例

Flink SQLの例

構成パラメータ

関連情報