リアルタイムデータベースインジェストジョブの構築 - Realtime Compute for Apache Flink

Realtime Compute for Apache Flink は、フル同期から増分同期への自動切り替え、メタデータディスカバリー、スキーマ進化、およびデータベース全体の同期を自動的に処理することで、リアルタイムデータインジェストを簡素化します。本トピックでは、ApsaraDB RDS for MySQL から Hologres へデータをストリーミングするデータインジェストジョブを迅速に構築する方法を説明します。

背景情報

次の図は、これらのデータベースとテーブルが DMS コンソールでどのように表示されるかを示しています。数据库和表情况

以下の手順に従って、すべてのテーブルを Hologres に同期し、シャード化されたユーザーテーブルを単一テーブルにマージするデータインジェストジョブを開発します。

ステップ 1：データインジェストジョブの開発
ステップ 2：ジョブの開始
ステップ 3：フル同期結果の検証
ステップ 4：増分同期結果の検証
（オプション）ステップ 5：ジョブリソースの構成

本トピックでは、Flink CDC データインジェストを使用して、データベース全体の同期とシャードテーブルのマージを実行します。これにより、1 つのジョブでフル同期、増分同期、およびリアルタイムスキーマ変更同期を完了できます。

前提条件

RAM ユーザーまたは RAM ロールを使用する場合は、Flink コンソールへのアクセスに必要な権限があることを確認してください。詳細については、「権限の管理」をご参照ください。
Flink ワークスペースを作成しました。詳細については、「Realtime Compute for Apache Flink をアクティブ化する」をご参照ください。
アップストリームおよびダウンストリームのストレージ
- ApsaraDB RDS for MySQL インスタンスを作成済みである必要があります。詳細については、「（非推奨、『ステップ 1』にリダイレクト）ApsaraDB RDS for MySQL インスタンスの迅速な作成」をご参照ください。
- Hologres インスタンスを作成済みである必要があります。詳細については、「Hologres インスタンスの購入」をご参照ください。
説明
ApsaraDB RDS for MySQL および Hologres のインスタンスは、Flink ワークスペースと同じリージョンおよび仮想プライベートクラウド (VPC) 内にある必要があります。そうでない場合は、ネットワーク接続を確立する必要があります。詳細については、「VPC 間で他のサービスにアクセスするには?」および「インターネットにアクセスするには?」をご参照ください。
テストデータを準備済みであり、IP ホワイトリストを構成済みである必要があります。詳細については、「MySQL テストデータおよび Hologres データベースの準備」および「IP ホワイトリストの構成」をご参照ください。

MySQL テストデータおよび Hologres データベースの準備

tpc_ds.sql、user_db1.sql、user_db2.sql、および user_db3.sql をクリックして、テストデータファイルをローカルマシンにダウンロードします。
DMS コンソールで、ご利用の ApsaraDB RDS for MySQL インスタンスにテストデータを準備します。
1. DMS を使用して、ご利用の ApsaraDB RDS for MySQL インスタンスにログインします。
  詳細については、「（非推奨、『ステップ 2』にリダイレクト）DMS を使用した ApsaraDB RDS for MySQL インスタンスへのログイン」をご参照ください。
2. SQL コンソールウィンドウで、次のコマンドを入力し、実行をクリックします。
  次のコマンドにより、tpc_ds、user_db1、user_db2、および user_db3 の 4 つのデータベースが作成されます。
```
CREATE DATABASE tpc_ds;
CREATE DATABASE user_db1;
CREATE DATABASE user_db2;
CREATE DATABASE user_db3;
```
3. 上部ナビゲーションバーで、データインポート をクリックします。
4. バッチデータインポート タブで、データをインポートするデータベースを選択し、対応する SQL ファイルをアップロードして、送信をクリックし、次に 変更の実行 をクリックします。表示されるダイアログボックスで、実行の確認 をクリックします。
  この操作を繰り返して、対応するデータファイルを tpc_ds、user_db1、user_db2、および user_db3 データベースにインポートします。
Hologres コンソールで、マージされたユーザーテーブルデータを格納するための my_user という名前のデータベースを作成します。
詳細については、「データベースの作成」をご参照ください。

IP ホワイトリストの構成

Flink が ApsaraDB RDS for MySQL および Hologres インスタンスにアクセスできるようにするには、ご利用の Flink ワークスペースの CIDR ブロックを両方のインスタンスの IP ホワイトリストに追加する必要があります。

Flink ワークスペースの CIDR ブロックを取得します。
1. Realtime Compute for Apache Flink コンソールにログインします。
2. ワークスペース一覧で、対象の ワークスペース を見つけ、操作列で その他 > ワークスペースの詳細 を選択します。
3. ワークスペースの詳細 ダイアログボックスで、Flink vSwitch の CIDR ブロック 情報を確認します。
Flink の CIDR ブロックを ApsaraDB RDS for MySQL インスタンスの IP ホワイトリストに追加します。
詳細については、「IP ホワイトリストの構成」をご参照ください。
Flink の CIDR ブロックを Hologres インスタンスの IP ホワイトリストに追加します。
HoloWeb でデータ接続を構成する際は、接続の IP ホワイトリストを構成する前に、ログイン方法 を 現在のユーザーによるパスワード不要ログイン に設定してください。詳細については、「IP ホワイトリスト」をご参照ください。

ステップ 1：インジェストジョブの開発

Flink 開発コンソールにログインし、新しいジョブを作成します。

データ開発 > データインジェスト ページで、新規作成 をクリックします。
空のデータインジェストドラフト をクリックします。
Realtime Compute for Apache Flink には、特定のユースケース、コードサンプル、およびガイダンスを備えた豊富なコードテンプレートが用意されています。テンプレートをクリックすると、プロダクト機能やビジネスロジックを実装するための構文について学ぶことができます。
次へをクリックします。

新規データインジェストジョブドラフト ダイアログボックスで、ジョブパラメーターを構成します。

パラメーター	説明	例
ファイル名	ジョブの名前です。説明ジョブ名は、現在のプロジェクト内で一意である必要があります。	flink-test
保存場所	ジョブのコードファイルを保存するフォルダです。既存のフォルダ横にあるアイコンをクリックして、サブフォルダを作成することもできます。	ジョブドラフト
エンジンバージョン	ジョブで使用する Flink エンジンバージョンです。エンジンバージョン番号、バージョン互換性、および重要なライフサイクル日付については、「エンジンバージョン」をご参照ください。	vvr-11.1-jdk11-flink-1.20

OK をクリックします。

次のジョブコードをジョブエディターにコピーします。

次のコードは、tpc_ds データベースのすべてのテーブルを Hologres に同期し、シャード化されたユーザーテーブルを Hologres の単一テーブルにマージします。

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: tpc_ds.\.*,user_db[0-9]+.user[0-9]+
  server-id: 8601-8604
  # (Optional) Synchronize table and column comments.
  include-comments.enabled: true
  # (Optional) Prioritize the distribution of unbounded chunks to prevent potential TaskManager OutOfMemory errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parsing filters to accelerate reading.
  scan.only.deserialize.captured.tables.changelog.enabled: true  

sink:
  type: hologres
  name: Hologres Sink
  endpoint: ****.hologres.aliyuncs.com:80
  dbname: cdcyaml_test
  username: ${secret_values.holo-username}
  password: ${secret_values.holo-password}
  sink.type-normalize-strategy: BROADEN
  
route:
  # Merge and synchronize the sharded user tables to the my_user.users table.
  - source-table: user_db[0-9]+.user[0-9]+
    sink-table: my_user.users

説明

MySQL の tpc_ds データベースのテーブルは、送信先で同じ名前のテーブルに直接マッピングされるため、route セクションに追加のマッピング構成は不要です。ods_tps_ds などの異なる名前のデータベースにテーブルを同期するには、次のように route モジュールを構成します。

route:
  # Merge and synchronize the sharded user tables to the my_user.users table.
  - source-table: user_db[0-9]+.user[0-9]+
    sink-table: my_user.users
  # Rename the database for all tables under tpc_ds and synchronize them to ods_tps_ds.
  - source-table: tpc_ds.\.*
    sink-table: ods_tps_ds.<>
    replace-symbol: <>

ステップ 2：ジョブの開始

データ開発 > データインジェスト ページで、デプロイ をクリックします。表示されるダイアログボックスで、確認をクリックします。
オペレーションセンター > ジョブ操作 ページで、対象ジョブの操作列の開始をクリックします。必要に応じてパラメーターを構成します。詳細については、「ジョブの開始」をご参照ください。
開始をクリックします。
ジョブが開始されると、ジョブ操作ページでランタイム情報とステータスを確認できます。

ステップ 3：フル同期の検証

Hologres 管理コンソールにログインします。
メタデータ管理 タブで、Hologres インスタンスの tpc_ds データベースに 24 個のテーブルとそのデータが存在することを確認します。
メタデータ管理 タブで、my_user データベースの users テーブルのスキーマを確認します。
次の図は、同期されたテーブルスキーマとデータを示しています。
- テーブルスキーマ
  users テーブルのスキーマには、ソースの MySQL テーブルには存在しない 2 つの追加カラム (_db_name および _table_name) が含まれています。これらのカラムは各行のソースデータベースおよびテーブルを示し、複合プライマリキーの一部として、シャードテーブルをマージした後のデータの一意性を保証します。
- テーブルデータ
  users テーブル情報ページの右上隅で、テーブルのクエリ をクリックします。次のコマンドを入力し、実行をクリックします。
```
select * from users order by _db_name,_table_name,id;
```
  クエリ結果を次の図に示します。

ステップ 4：増分同期の検証

フル同期が完了すると、ジョブは手動介入なしに自動的に増分同期フェーズに切り替わります。「モニタリングとアラート」タブの currentEmitEventTimeLag 値を確認することで、データ同期フェーズを判断できます。

Realtime Compute for Apache Flink コンソールにログインします。
対象ワークスペースの操作列で、コンソール をクリックします。
オペレーションセンター > ジョブ操作 ページで、対象ジョブの名前をクリックします。
モニタリングとアラート（または メトリクス）タブをクリックします。
currentEmitEventTimeLag グラフを確認して、データ同期フェーズを判断します。
- 値が 0 の場合、フル同期フェーズです。
- 値が 0 より大きい場合、増分同期フェーズです。
リアルタイムデータおよびスキーマ変更同期を検証します。
MySQL CDC ソースは、増分フェーズ中にリアルタイムデータおよびスキーマ同期をサポートします。これを検証するには、ジョブがこのフェーズに入った後に、MySQL のシャードユーザーテーブルのスキーマおよびデータを変更します。
1. DMS を使用して ApsaraDB RDS for MySQL インスタンスにログインします。
  詳細については、「（非推奨、『ステップ 2』にリダイレクト）DMS を使用した ApsaraDB RDS for MySQL インスタンスへのログイン」をご参照ください。
2. user_db2 データベースで、次のコマンドを実行して user02 テーブルのスキーマを変更し、データを挿入および更新します。
```
USE `user_db2`;
ALTER TABLE `user02` ADD COLUMN `age` INT;   -- Add the age column.
INSERT INTO `user02` (id, name, age) VALUES (27, 'Tony', 30); -- Insert a row that includes the age data.
UPDATE `user05` SET name='JARK' WHERE id=15;  -- Update another table and change the name to uppercase.
```
3. Hologres コンソールで、users テーブルのスキーマおよびデータの変更を確認します。
  users テーブル情報ページの右上隅で、テーブルのクエリ をクリックし、次のコマンドを入力して、実行をクリックします。
```
select * from users order by _db_name,_table_name,id;
```
  次の図はクエリ結果を示しています。user02 のスキーマ変更およびデータ修正がリアルタイムで伝播しており、シャードテーブルのスキーマが異なっていても問題ありません。Hologres の users テーブルには、新しい age カラム、Tony の挿入レコード、および JARK の更新レコードが表示されています。

（オプション）ステップ 5：ジョブリソースの構成

パフォーマンスを向上させるために、データ量に応じて同時実行数、TaskManager メモリ、CU などのジョブリソースを調整できます。

オペレーションセンター > ジョブ操作 ページで、対象ジョブの名前をクリックします。
デプロイメントの詳細 タブで、リソース構成 セクションの右上隅にある編集をクリックします。
TaskManager メモリや同時実行数などのリソースパラメーターを手動で設定します。
リソース構成 セクションの右側で、保存をクリックします。
ジョブを再起動します。
リソース構成の変更は、ジョブを再起動した後でのみ有効になります。

Realtime Compute for Apache Flink:リアルタイムデータベースインジェスト