Data Transmission Service:自己管理 MySQL から Message Queue for Apache Kafka へのデータ同期

注意事項

• 初期完全データ同期中、DTS はソースデータベースとターゲットデータベースの読み書きリソースを消費するため、データベースの負荷が増加します。データベースパフォーマンスが低い、インスタンスの仕様が低い、またはビジネストラフィックが重い (たとえば、ソースデータベースに多くの低速 SQL クエリやプライマリキーのないテーブルがある、またはターゲットデータベースでデッドロックが発生するなど) 場合、データベースの負荷が増加し、サービスが利用できなくなる可能性さえあります。データを同期する前に、ソースインスタンスと宛先インスタンスのパフォーマンスを評価してください。両方のインスタンスの CPU 使用率が 30% 未満であるなど、オフピーク時にデータ同期を実行することを推奨します。

• ソーステーブルにプライマリキーまたは一意制約がなく、その列の組み合わせが一意でない場合、重複データがターゲットデータベースに書き込まれる可能性があります。

課金

制限事項

• テーブルのみ同期可能です。他のオブジェクトタイプはサポートされていません。
• 同期オブジェクトの自動調整はサポートされていません。同期されたテーブルの名前を変更し、新しい名前が同期オブジェクトのリストに含まれていない場合、そのテーブルからのデータは宛先の Kafka クラスターに同期されなくなります。名前が変更されたテーブルの同期を続行するには、オブジェクトの再選択 を行う必要があります。詳細については、「同期オブジェクトの追加」をご参照ください。

事前準備

操作手順

1. データ同期インスタンスを購入します。詳細については、「DTS インスタンスの購入」をご参照ください。
   説明 インスタンスを購入する際、ソースインスタンスを MySQL、宛先インスタンスを Kafka、同期トポロジを 一方向同期 に設定します。

2. DTS コンソールにログインします。

   説明

   自動的に Data Management (DMS) コンソールにリダイレクトされた場合は、右下隅の アイコンをクリックし、次に をクリックして従来の DTS コンソールに戻ることができます。

3. 左側のナビゲーションウィンドウで、データ同期 をクリックします。

4. [同期タスク] ページの上部で、宛先インスタンスが配置されているリージョンを選択します。

5. 購入したデータ同期タスクを見つけ、[タスクの設定] をクリックします。

6. データ同期タスクのソースインスタンスと宛先インスタンスを設定します。

   セクション	パラメーター	説明
   N/A	同期タスク名	DTS は自動的にタスク名を生成します。簡単に識別できるよう、分かりやすい名前を指定することを推奨します。名前は一意である必要はありません。
   ソースインスタンス情報	インスタンスタイプ	ソースデータベースがデプロイされている場所に基づいてインスタンスタイプを選択します。この例では、[ECS インスタンス上の自己管理データベース] が選択されています。他のインスタンスタイプの設定も同様です。
   	インスタンスリージョン	購入時に選択したソースインスタンスのリージョンです。このパラメーターは変更できません。
   	ECS インスタンス ID	自己管理 MySQL データベースをホストする Elastic Compute Service (ECS) インスタンスの ID を選択します。
   	データベースタイプ	このパラメーターは MySQL に固定されており、変更できません。
   	ポート	自己管理 MySQL データベースのサービスポートを入力します。
   	データベースアカウント	自己管理 MySQL データベースのアカウントのユーザー名を入力します。アカウントには、同期するすべてのオブジェクトに対する REPLICATION CLIENT、REPLICATION SLAVE、SHOW VIEW、および SELECT 権限が必要です。
   	データベースパスワード	データベースアカウントのパスワードを入力します。
   宛先インスタンス情報	インスタンスタイプ	[パブリック IP、Express Connect、VPN Gateway、または Smart Access Gateway を使用した自己管理データベース] を選択します。 説明 現在、DTS は Message Queue for Apache Kafka を直接選択することをサポートしていません。回避策として、自己管理 Kafka インスタンスとして設定します。
   	インスタンスリージョン	購入時に選択した宛先インスタンスのリージョンです。このパラメーターは変更できません。
   	ピア VPC	宛先 Kafka インスタンスが配置されている VPC を選択します。VPC ID は、ご利用の Kafka インスタンスの 基本情報 ページで確認できます。
   	データベースタイプ	Kafka を選択します。
   	IP アドレス	Kafka インスタンスの [デフォルトエンドポイント] から任意の IP アドレスを入力します。 説明 IP アドレスは、ご利用の Kafka インスタンスの 基本情報 ページの [デフォルトエンドポイント] フィールドから取得できます。
   	ポート	Kafka インスタンスのサービスポートです。デフォルト値：9092。
   	データベースアカウント	Kafka インスタンスのユーザー名を入力します。 説明 Kafka インスタンスのインスタンスタイプが [VPC インスタンス] の場合、データベースアカウント と データベースのパスワード パラメーターを設定する必要はありません。
   	データベースパスワード	ユーザー名のパスワードを入力します。
   	トピック	右側の [トピックリストの取得] をクリックし、ドロップダウンリストからトピックを選択します。
   	Kafka バージョン	ご利用の Kafka インスタンスのバージョンを選択します。
   	接続方法	ビジネスおよびセキュリティ要件に基づいて、非暗号化 または SCRAM-SHA-256 を選択します。

7. ページ右下隅の [ホワイトリストを設定して次へ] をクリックします。

   ソースまたはターゲットデータベースが ApsaraDB RDS for MySQL や ApsaraDB for MongoDB などの Alibaba Cloud データベースインスタンスである場合、DTS は DTS サーバーの CIDR ブロックをインスタンスの IP アドレスホワイトリストに自動的に追加します。ソースまたはターゲットデータベースが ECS インスタンスでホストされている自己管理データベースである場合、DTS は DTS サーバーの CIDR ブロックを ECS インスタンスのセキュリティグループルールに自動的に追加します。この場合、ECS インスタンスがデータベースにアクセスできることを確認する必要があります。自己管理データベースが複数の ECS インスタンスでホストされている場合は、各 ECS インスタンスのセキュリティグループルールに DTS サーバーの CIDR ブロックを手動で追加する必要があります。ソースまたはターゲットデータベースがデータセンターにデプロイされているか、サードパーティのクラウドサービスプロバイダーによって提供されている自己管理データベースである場合は、DTS がデータベースにアクセスできるように、DTS サーバーの CIDR ブロックをデータベースの IP アドレスホワイトリストに手動で追加する必要があります。詳細については、「DTS サーバーの IP アドレスのホワイトリスト登録」をご参照ください。

   警告

   DTS サーバーのパブリック IP アドレスの CIDR ブロックを自動または手動で追加すると、セキュリティリスクが生じる可能性があります。このプロダクトを使用することにより、これらの潜在的なリスクを認識し、受け入れるものとします。基本的なセキュリティ保護を実装する必要があります。これらの保護には、パスワードセキュリティの強化、CIDR ブロックのオープンポートの制限、内部 API 通信での認証の使用、不要な CIDR ブロックの定期的な確認と制限などが含まれますが、これらに限定されません。または、専用線、VPN Gateway、Smart Access Gateway などの内部ネットワーク経由でデータベースに接続することもできます。

8. 同期オブジェクトを設定します。

   パラメーター	説明
   Kafka のデータ形式	Kafka クラスターに同期されたデータは、Avro または Canal JSON 形式で保存されます。詳細については、「メッセージキューのデータ形式」をご参照ください。
   [Kafka パーティションへのデータ転送ポリシー]	ビジネス要件に合ったポリシーを選択します。詳細な説明については、「Kafka パーティションへのデータ同期ポリシー」をご参照ください。
   同期オブジェクト	ソースオブジェクト ボックスで、同期するオブジェクト (テーブルが最も細かい粒度) を選択し、 アイコンをクリックして [選択済み] ボックスに移動します。 説明 DTS は、テーブル名をステップ 6 で選択したトピック名に自動的にマッピングします。テーブルの宛先トピックを変更するには、オブジェクト名マッピング機能を使用します。詳細については、「宛先インスタンスでのオブジェクト名の設定。
   オブジェクト名マッピング	宛先インスタンスで同期されたオブジェクトの名前を変更します。詳細については、「データベース、テーブル、および列のマッピング」をご参照ください。
   接続失敗時のリトライ時間	DTS がソースまたは宛先インスタンスに接続できない場合、デフォルトで 720 分 (12 時間) リトライします。カスタムのリトライ持続時間を指定することもできます。指定された持続時間内に DTS がソースまたは宛先インスタンスに再接続した場合、同期タスクは自動的に再開されます。そうでない場合、タスクは失敗します。 説明 接続リトライ中のタスク実行時間に対して課金されます。ビジネスニーズに基づいてリトライ持続時間をカスタマイズするか、ソースおよび宛先インスタンスがリリースされるとすぐに DTS インスタンスをリリースしてください。

9. 上記の設定を完了した後、ページ右下隅の 次へ をクリックします。
10. 初期同期の詳細設定を構成します。

    パラメーター	説明
    [初期同期]	デフォルトでは、[初期スキーマ同期] と [初期完全データ同期] の両方が選択されています。増分データを同期する前に、DTS は選択したオブジェクトのスキーマと既存のデータを宛先に同期します。
    [フィルターオプション]	デフォルトでは、[増分同期フェーズで DDL を無視] が選択されています。これは、増分データ同期中にソースデータベースで実行された DDL 操作を DTS が同期しないことを意味します。

11. 上記の設定を完了した後、ページ右下隅の [事前チェックして開始] をクリックします。

    説明

    • 同期タスクが開始される前に事前チェックが実行され、事前チェックに合格した後にのみタスクを開始できます。

    • 事前チェックが失敗した場合は、失敗した項目の横にある アイコンをクリックして詳細を表示します。

      • 原因に基づいて問題を修正し、再度事前チェックを実行できます。

      • 警告をトリガーした項目を修正する必要がない場合は、無視 または [警告を無視して事前チェックを再実行] をクリックして警告をスキップし、再度事前チェックを実行できます。

12. 事前チェック ダイアログボックスに 事前チェック完了 と表示されたら、事前チェック ダイアログボックスを閉じます。データ同期タスクが開始されます。
    データ同期 ページでは、タスクリストにインスタンス ID/タスク名、ステータス (例：同期中)、同期の概要 (レイテンシと速度を含む)、課金方法、同期トポロジなどの主要情報が表示されます。操作列には、同期の一時停止、サブスクリプションへの変換、アップグレードなどのオプションがあります。