Realtime Compute for Apache Flink:OceanBase

一般パラメーター

これらのパラメーターは、すべてのテーブルタイプに適用されます。

ソーステーブルのパラメーター

ディメンションテーブルパラメーター

シンクテーブルパラメーター (JDBC)

シンクテーブルのパラメータ（バイパスインポート）

バイパスインポートは、OceanBase でのバルクデータロードのための高スループットな書き込み方法です。VVR 11.5 以降で利用できます。

バイパスインポートを使用する前に、以下の制約を確認してください。

• 有界ストリームのみ：データソースは有界ストリームである必要があります。最高のパフォーマンスを得るには、Flink バッチモードを使用してください。

• インポート中のテーブルロック：ターゲットテーブルは、インポート全体を通じてロックされます。DML 書き込みと DDL 変更はブロックされますが、読み取りクエリは影響を受けません。

• リアルタイム書き込みには不適：ストリーミングまたはリアルタイム書き込みの場合は、代わりに JDBC シンクを使用してください。

MySQL 互換モード

Oracle 互換モード

ソーステーブルとシンクテーブル

次の例では、OceanBase のソーステーブルから CDC データを読み取り、JDBC シンクテーブルに書き込みます。参考として、バイパスインポートのシンクテーブル定義も示します。

3 つのテーブルはいずれも 'connector' = 'oceanbase' を使用します。ソースと JDBC シンクでは使用するパラメーターセットが異なります。ダイレクトロードのシンクでは sink.mode = 'direct-load' を設定し、RPC ポート経由で接続します。

-- OceanBase CDC ソーステーブル (全履歴を読み取った後、Binlog を読み取ります)
CREATE TEMPORARY TABLE oceanbase_source (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector'     = 'oceanbase',
  'hostname'      = '<your-hostname>',
  'port'          = '3306',
  'username'      = '<your-username>',
  'password'      = '<your-password>',
  'database-name' = '<your-database-name>',
  'table-name'    = '<your-table-name>'
);

-- OceanBase JDBC シンクテーブル (リアルタイムのストリーミング書き込み用)
CREATE TEMPORARY TABLE oceanbase_sink (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'oceanbase',
  'url'       = '<your-jdbc-url>',
  'userName'  = '<your-username>',
  'password'  = '<your-password>',
  'tableName' = '<your-table-name>'
);

-- OceanBase バイパスインポートのシンクテーブル (高スループットなバッチ書き込み用)
-- 有界データソースが必要です。最適なパフォーマンスを得るには、Flink をバッチモードに設定してください
CREATE TEMPORARY TABLE oceanbase_directload_sink (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector'   = 'oceanbase',
  'sink.mode'   = 'direct-load',
  'host'        = '<your-host>',
  'port'        = '<your-rpc-port>',
  'tenant-name' = '<your-tenant-name>',
  'schema-name' = '<your-schema-name>',
  'table-name'  = '<your-table-name>',
  'username'    = '<your-username>',
  'password'    = '<your-password>'
);

BEGIN STATEMENT SET;
INSERT INTO oceanbase_sink
SELECT * FROM oceanbase_source;
END;

ディメンションテーブル

次の例では、Datagen ソースを OceanBase のディメンションテーブルとテンポラル結合します。ALL キャッシュポリシーは、ジョブ開始前にディメンションテーブル全体をメモリにロードします。

CREATE TEMPORARY TABLE datagen_source (
  a INT,
  b BIGINT,
  c STRING,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

-- ALL キャッシュポリシーを使用する OceanBase ディメンションテーブル
-- ALL キャッシュは起動時にテーブル全体をロードします。小規模で安定したテーブルに適しています
CREATE TEMPORARY TABLE oceanbase_dim (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'oceanbase',
  'url'       = '<your-jdbc-url>',
  'userName'  = '<your-username>',
  'password'  = '${secret_values.password}',
  'tableName' = '<your-table-name>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  a INT,
  b STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT T.a, H.b
FROM datagen_source AS T
JOIN oceanbase_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H
ON T.a = H.a;