AnalyticDB:Flink によるバイナリログのサブスクライブ

前提条件

• AnalyticDB for MySQL クラスターは、Enterprise Edition、Basic Edition、Data Lakehouse Edition、または エラスティックモードの Data Warehouse Edition です。

• AnalyticDB for MySQL クラスターのマイナーバージョンは 3.2.1.0 以降です。

  説明

  AnalyticDB for MySQL クラスターのマイナーバージョンを表示および更新するには、AnalyticDB for MySQL コンソールにログインし、クラスター情報 ページの 構成情報 セクションに移動します。

• Flink ワークスペースは、Ververica Runtime (VVR) 8.0.4 またはそれ以降のバージョンを使用する必要があります。

• AnalyticDB for MySQL クラスターとフルマネージド Flink ワークスペースは、同じ VPC 内にある必要があります。

• Flink ワークスペースの CIDR ブロックが、AnalyticDB for MySQL のホワイトリストに追加されている必要があります。

制限事項

• Flink は、AnalyticDB for MySQL のバイナリログから、基本データ型と JSON 複合データ型のみを処理できます。

• Flink を使用して AnalyticDB for MySQL のバイナリログをサブスクライブする場合、次の操作はキャプチャされません。対応するデータ変更はダウンストリームコンシューマーに同期されません。

  • DDL 操作：ALTER TABLE などのスキーマ変更はキャプチャされません。ソーステーブルスキーマが変更された場合は、Flink ジョブ内のテーブル定義を手動で更新し、ジョブを再デプロイする必要があります。

  • INSERT OVERWRITE：INSERT OVERWRITE によって書き込まれたデータでは、バイナリログレコードが生成されません。INSERT OVERWRITE 操作後の完全なデータを取得するには、バイナリログを再サブスクライブする必要があります。

  • 自動パーティションエビクション：パーティションテーブル内の期限切れのパーティションが自動的にエビクションされる場合、エビクションされたデータではバイナリログレコードが生成されません。

  • マテリアライズドビューのフルリフレッシュ：マテリアライズドビューに対するフルリフレッシュ操作では、バイナリログレコードが生成されません。

    説明

    マテリアライズドビューのバイナリログには、インクリメンタルリフレッシュによるデータ変更のみが含まれます。フルリフレッシュ後の完全なデータ状態を取得するには、バイナリログを再サブスクライブする必要があります。

ステップ3：バイナリログのサブスクリプション

1. Realtime Compute for Apache Flink コンソールにログインし、SQL ジョブを作成します。

2. AnalyticDB for MySQL に接続し、特定のテーブル (source_table) からバイナリログデータを読み取るソーステーブルを作成します。

   説明

   • Flink DDL で定義されたプライマリキーは、AnalyticDB for MySQL クラスターの物理テーブルのプライマリキーと一致する必要があります。これには、キー列とプライマリキー名の両方が含まれます。一致しない場合、データが不正確になる可能性があります。

   • Flink のデータ型は、AnalyticDB for MySQL のデータ型と互換性がある必要があります。データ型のマッピングの詳細については、「型マッピング」をご参照ください。

   CREATE TEMPORARY TABLE adb_source (
     `id` INT,
     `num` BIGINT,
     PRIMARY KEY (`id`) NOT ENFORCED
   ) WITH (
     'connector' = 'adb-mysql-cdc',
     'hostname' = 'amv-2zepb9n1l58ct01z50000****.ads.aliyuncs.com',
     'username' = 'testUser',
     'password' = 'Test12****',
     'database-name' = 'binlog',
     'table-name' = 'source_table'
   );

   次の表では、WITH パラメーターについて説明します。

   パラメーター	必須	デフォルト	型	説明
   connector	はい	なし	STRING	使用するコネクタ。 これはカスタムコネクタです。このパラメーターを adb-mysql-cdc に設定します。
   hostname	はい	なし	STRING	AnalyticDB for MySQL クラスターのエンドポイント。
   username	はい	なし	STRING	AnalyticDB for MySQL クラスターのデータベースアカウント。
   password	はい	なし	STRING	AnalyticDB for MySQL データベースアカウントのパスワード。
   database-name	はい	なし	STRING	AnalyticDB for MySQL データベースの名前。 AnalyticDB for MySQL はテーブルレベルのバイナリロギングを実装しているため、指定できるデータベースは 1 つだけです。
   table-name	はい	なし	STRING	AnalyticDB for MySQL データベースのテーブルの名前。 AnalyticDB for MySQL はテーブルレベルのバイナリロギングを実装しているため、指定できるテーブルは 1 つだけです。
   port	いいえ	3306	INTEGER	ポート番号。
   scan.incremental.snapshot.enabled	いいえ	true	BOOLEAN	インクリメンタルスナップショット 読み取りメカニズムを有効にするかどうかを指定します。 この機能はデフォルトで有効になっています。インクリメンタルスナップショットは、テーブルスナップショットを読み取るための新しいメカニズムです。以前のスナップショットメカニズムと比較して、インクリメンタルスナップショットには次の利点があります： ソースはスナップショット読み取り中に同時読み取りをサポートします。 ソースはスナップショット読み取り中にチャンクレベルのチェックポイントをサポートします。 ソースはスナップショットを読み取る前にデータベースロック権限を取得する必要はありません。
   scan.incremental.snapshot.chunk.size	いいえ	8096	INTEGER	テーブルスナップショットのチャンクあたりの行数。
   scan.snapshot.fetch.size	いいえ	1024	INTEGER	テーブルスナップショットを読み取るときに一度にフェッチする最大行数。
   scan.startup.mode	いいえ	initial	STRING	データ読み取りの起動モード。 有効な値： initial (デフォルト)：テーブルの初期スナップショットを実行し、最新のバイナリログを読み取ります。 earliest-offset：スナップショットフェーズをスキップし、利用可能な最も古いバイナリログから読み取りを開始します。 specific-offset：スナップショットフェーズをスキップし、指定されたバイナリログの位置から開始します。scan.startup.specific-offset.file と scan.startup.specific-offset.pos パラメーターの両方を設定して、バイナリログのファイル名とオフセットを指定します。 latest-offset：スナップショットフェーズをスキップし、コネクタの起動後に発生した変更のみを読み取ります。 timestamp：スナップショットフェーズをスキップし、指定されたタイムスタンプから読み取りを開始します。scan.startup.timestamp-millis パラメーターを使用して、タイムスタンプをミリ秒 (ms) 単位で設定します。 重要 earliest-offset、specific-offset、または timestamp 起動モードを使用する場合は、指定された開始位置とジョブの起動の間でテーブルスキーマが変更されないようにしてください。変更されると、スキーマの変更が原因でジョブが失敗する可能性があります。
   scan.startup.specific-offset.file	いいえ	なし	STRING	specific-offset 起動モードでは、開始位置のバイナリログファイル名。 最新のバイナリログファイル名を取得するには、SHOW MASTER STATUS for table_name; ステートメントを実行してください。
   scan.startup.specific-offset.pos	いいえ	なし	LONG	specific-offset 起動モードでは、開始位置のバイナリログファイル内の位置。 最新のバイナリログの位置を取得するには、SHOW MASTER STATUS for table_name; ステートメントを実行してください。
   scan.startup.specific-offset.skip-events	いいえ	なし	LONG	指定された開始位置の後にスキップするイベントの数。
   scan.startup.specific-offset.skip-rows	いいえ	なし	LONG	指定された開始位置の後にスキップするデータ行の数。
   scan.startup.timestamp-millis	いいえ	なし	LONG	timestamp 起動モードを使用する場合の、開始タイムスタンプ (ミリ秒)。 このパラメーターを使用する場合は、scan.startup.mode を timestamp に設定する必要があります。タイムスタンプはミリ秒 (ms) 単位です。
   server-time-zone	いいえ	システムのデフォルト	STRING	データベースサーバーのセッションタイムゾーン (例："Asia/Shanghai")。 これは、AnalyticDB for MySQL の TIMESTAMP 型が STRING データ型に変換される方法を制御します。このパラメーターが設定されていない場合、ZoneId.systemDefault() がサーバーのタイムゾーンを決定するために使用されます。
   debezium.min.row.count.to.stream.result	いいえ	1000	INTEGER	テーブルの行数がこの値より大きい場合、コネクタは結果をストリーミングします。 このパラメーターを 0 に設定すると、すべてのテーブルサイズチェックがスキップされ、スナップショットフェーズですべての結果がストリーミングされます。
   connect.timeout	いいえ	30s	DURATION	データベースへの接続試行がタイムアウトするまでの最大待機時間。 デフォルトの単位は秒 (s) です。
   connect.max-retries	いいえ	3	INTEGER	データベース接続失敗後の最大再試行回数。

3. 宛先データベースに物理テーブルを作成して、処理されたデータを保存します。このトピックでは、宛先として AnalyticDB for MySQL を使用します。Flink でサポートされているコネクタについては、「サポートされているコネクタ」をご参照ください。

   CREATE TABLE target_table (
     `id` INT,
     `num` BIGINT,
     PRIMARY KEY (`id`)
   )

4. 前のステップで作成したテーブルに接続するシンクテーブルを作成します。シンクテーブルは、処理されたデータを AnalyticDB for MySQL の指定されたテーブルに書き込みます。

   CREATE TEMPORARY TABLE adb_sink (
     `id` INT,
     `num` BIGINT,
     PRIMARY KEY (`id`) NOT ENFORCED
   ) WITH (
     'connector' = 'adb3.0',
     'url' = 'jdbc:mysql://amv-2zepb9n1l58ct01z50000****.ads.aliyuncs.com:3306/flinktest',
     'userName' = 'testUser',
     'password' = 'Test12****',
     'tableName' = 'target_table'
   );

   シンクテーブルの WITH パラメーターと型マッピングの詳細については、「AnalyticDB for MySQL V3.0 コネクタ」をご参照ください。

5. INSERT INTO ステートメントを使用して、ソーステーブルからシンクテーブルにデータを送信します。

   INSERT INTO adb_sink
   SELECT * FROM adb_source;

6. 保存 をクリックします。

7. [Validate] をクリックします。

   検証機能は、ジョブで使用されるテーブルの SQL セマンティクス、ネットワーク接続性、メタデータをチェックします。結果エリアで [SQL Advice] をクリックして、SQL のリスクプロンプトや最適化の提案を表示することもできます。

8. (任意) Debug をクリックします。

   ジョブデバッグ機能を使用して、ジョブの実行をシミュレートし、出力結果を確認し、SELECT または INSERT ステートメントのビジネスロジックを検証し、開発効率を向上させ、データ品質のリスクを低減できます。

9. Deploy をクリックします。

   ジョブを開発および検証した後、本番環境にデプロイします。その後、[O&M] ページに移動し、ジョブを開始します。

10. (オプション) バイナリログに関する情報を表示します。

    説明

    次の SQL 文を実行してバイナリログに関する情報をクエリした後、バイナリログ機能を有効にしただけの場合は 0 が返されます。ログ情報は、バイナリログをサブスクライブした後にのみ表示されます。

    • 最新のバイナリログのファイル名と場所をクエリするには、次の文を実行します。

      SHOW MASTER STATUS FOR source_table;

    • クリアされていないすべての履歴バイナリログとそのサイズをクエリするには、次の文を実行します。

      SHOW BINARY LOGS FOR source_table;

型マッピング

次の表は、AnalyticDB for MySQL のデータ型と、それに対応する Flink のデータ型を示しています。