AnalyticDB:Realtime Compute for Apache Flink を使用した AnalyticDB for PostgreSQL へのデータ書き込み

制限事項

• Realtime Compute for Apache Flink は、サーバーレスモードの AnalyticDB for PostgreSQL からのデータ読み取りをサポートしていません。

• AnalyticDB for PostgreSQL コネクタは、Ververica Runtime (VVR) 6.0.0 以降でのみサポートされます。

• AnalyticDB for PostgreSQL V7.0 は、VVR 8.0.1 以降でのみサポートされます。

前提条件

開始する前に、以下の条件を満たしていることを確認してください。

• Flink 完全管理ワークスペースが作成済みであること。詳細については、「Flink 完全管理の有効化」をご参照ください。

• AnalyticDB for PostgreSQL インスタンスが作成済みであること。詳細については、「インスタンスの作成」をご参照ください。

• AnalyticDB for PostgreSQL インスタンスと Flink 完全管理ワークスペースが、同一の仮想プライベートクラウド (VPC) 内に配置されていること。

AnalyticDB for PostgreSQL インスタンスの設定

1. AnalyticDB for PostgreSQL コンソールにログインします。

2. Flink 完全管理ワークスペースの CIDR ブロックを、インスタンスの IP アドレスホワイトリストに追加します。詳細については、「IP アドレスホワイトリストの設定」をご参照ください。

3. データベースへのログイン をクリックします。接続オプションについては、「クライアント接続」をご参照ください。

4. 送信先テーブルを作成します。

   CREATE TABLE test_adbpg_table(
     b1 int,
     b2 int,
     b3 text,
     PRIMARY KEY(b1)
   );

Flink コネクタのセットアップ

AnalyticDB for PostgreSQL コネクタは、GitHub でホストされる JAR ファイルとして配布されています。このファイルをダウンロードし、Flink ワークスペースにカスタムコネクタとしてアップロードしてください。

1. Realtime Compute for Apache Flink コンソールにログインします。Flink 完全管理 タブで、対象のワークスペースを見つけ、コンソール を 操作 列からクリックします。

2. 左側ナビゲーションウィンドウで、コネクタ をクリックします。

3. カスタムコネクタの作成 をクリックし、JAR ファイルをアップロードします。

   説明 JAR ファイルは、「GitHub リリース」から取得してください。JAR のバージョンは、ワークスペースの Flink エンジンバージョンと一致させる必要があります。

4. 次へ をクリックします。システムが JAR の内容を解析します。解析に失敗した場合は、コネクタコードが Apache Flink コミュニティの標準に準拠しているか確認してください。

5. 完了 をクリックします。コネクタがコネクタ一覧に表示されます。

Flink ジョブの作成

1. Realtime Compute for Apache Flink コンソールにログインします。Flink 完全管理 タブで、対象のワークスペースを見つけ、コンソール を 操作 列からクリックします。

2. 左側ナビゲーションウィンドウで、SQL エディタ をクリックします。左上隅の 新規 をクリックします。

3. 新規ドラフト ダイアログボックスで、SQL スクリプト タブを選択し、空白ストリームドラフト をクリックして 次へ をクリックします。

4. ドラフトを設定し、作成 をクリックします。

   パラメーター	説明	例
   名前	現在のプロジェクト内で一意となるドラフト名	adbpg-test
   場所	コードファイルの保存先フォルダ	Draft
   エンジンバージョン	Flink エンジンバージョン。サポートされるバージョンについては、「エンジンバージョン」をご参照ください。	vvr-6.0.7-flink-1.15

AnalyticDB for PostgreSQL へのデータ書き込み

-- ソース：組み込み datagen コネクタを用いたランダムデータ生成
CREATE TABLE datagen_source (
  f_sequence INT,
  f_random INT,
  f_random_str STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '5',
  'fields.f_sequence.kind' = 'sequence',
  'fields.f_sequence.start' = '1',
  'fields.f_sequence.end' = '1000',
  'fields.f_random.min' = '1',
  'fields.f_random.max' = '1000',
  'fields.f_random_str.length' = '10'
);

-- 結果：adbpg コネクタを経由した AnalyticDB for PostgreSQL へのデータ書き込み
CREATE TABLE test_adbpg_table (
  `B1` bigint,
  `B2` bigint,
  `B3` VARCHAR,
  `B4` VARCHAR,
  PRIMARY KEY(B1) NOT ENFORCED
) WITH (
  'connector' = 'adbpg-nightly-1.13',
  'url' = 'jdbc:postgresql://<internal-endpoint>:5432/<database-name>',
  'tablename' = 'test_adbpg_table',
  'username' = '<username>',
  'password' = '<password>',
  'maxretrytimes' = '2',
  'batchsize' = '50000',
  'connectionmaxactive' = '5',
  'conflictmode' = 'ignore',
  'usecopy' = '0',
  'targetschema' = 'public',
  'exceptionmode' = 'ignore',
  'casesensitive' = '0',
  'writemode' = '1',
  'retrywaittime' = '200'
);

結果の検証

1. AnalyticDB for PostgreSQL データベースに接続します。詳細については、「クライアント接続」をご参照ください。

2. テーブルをクエリして、データが書き込まれていることを確認します。

   SELECT * FROM test_adbpg_table;

   データは AnalyticDB for PostgreSQL データベースに書き込まれます。次の図に返された結果を示します。

   

次のステップ

• AnalyticDB for PostgreSQL コネクタ — コネクタの完全なリファレンス（すべてのパラメーターおよびデータ型マッピングを含む）

• Datagen コネクタ — この例で使用される datagen ソースのリファレンス

参照情報

• Datagen