Data Transmission Service:ApsaraDB RDS for MySQL インスタンスから ApsaraMQ for Kafka インスタンスへのデータ同期

ソースデータベース

• テーブルにはプライマリキーまたはすべてのフィールドが一意となる一意制約が必要です。これがない場合、宛先に重複レコードが含まれる可能性があります。

• 個別のテーブル（データベース全体ではなく）を選択し、同期中にテーブル名またはカラム名を変更する場合、1 つのタスクで最大 1,000 のテーブルをサポートします。1,000 を超えるテーブルの場合は、複数のタスクを構成するか、代わりにデータベース全体を同期してください。

• バイナリログの要件：

  • binlog_row_image を full に設定してください。このパラメーターが正しく設定されていない場合、事前チェックが失敗し、タスクを開始できません。

  • ApsaraDB RDS for MySQL では、バイナリログを少なくとも 3 日間保持してください（7 日間推奨）。セルフマネージド MySQL の場合は、ログを少なくとも 7 日間保持してください。保持期間が短いと、タスクの失敗やデータ損失が発生する可能性があり、サービスレベル契約 (SLA) に基づく DTS サービスの信頼性に影響を与える可能性があります。詳細については、「バイナリログファイルの削除」をご参照ください。

  • セルフマネージド MySQL の場合、binlog_format も row に設定します。デュアルプライマリクラスターでは、DTS がすべてのバイナリログを取得できるように、log_slave_updates を ON に設定します。詳細については、「セルフマネージド MySQL データベースのアカウントを作成し、バイナリロギングを設定する」をご参照ください。

• スキーマ同期または完全データ同期中に、データベースまたはテーブルスキーマを変更する DDL ステートメントを実行しないでください。これによりタスクが失敗します。

• バイナリログ変更操作（物理バックアップから復元されたデータやカスケード操作からのデータなど）によって生成されたデータは、キャプチャまたは同期されません。必要に応じて、影響を受けるデータベースとテーブルを同期オブジェクトから削除して、再度追加できます。詳細については、「同期するオブジェクトの変更」をご参照ください。

• MySQL 8.0.23 以降では、不可視カラムは同期できず、そのデータは失われます。カラムを可視化するには、ALTER TABLE <table_name> ALTER COLUMN <column_name> SET VISIBLE; を実行してください。明示的なプライマリキーを持たないテーブルは、不可視のプライマリキーを自動生成する場合があります。それらも可視化してください。「不可視カラム」および「生成された不可視プライマリキー」をご参照ください。

• 読み取り専用の ApsaraDB RDS for MySQL 5.6 インスタンスは、トランザクションログを記録しないため、ソースとして使用できません。

その他の制限事項

• DTS は外部キーを同期しません。ソースでトリガーされるカスケード操作および削除操作は、宛先に伝播されません。

• 完全データ同期では、ソースおよび宛先インスタンスの読み取り・書き込みリソースを使用するため、データベース負荷が増加します。可能であれば、オフピーク時間帯に同期を実行してください。

• 完全データ同期中、同時実行の INSERT 操作により、宛先でテーブルの断片化が発生します。完全同期後、宛先の表領域は通常、ソースよりも大きくなります。

• 同期対象のオブジェクトとしてデータベース全体ではなく 1 つ以上のテーブルを選択した場合、同期中に pt-online-schema-change などのツールを使用してそれらのテーブルに対してオンライン DDL 操作を実行しないでください。これにより同期が失敗する可能性があります。代わりに、オンライン DDL には Data Management (DMS) を使用してください。『ロックフリー DDL 操作を実行する』をご参照ください。

• 同期中に他のソースから宛先にデータを書き込まないでください。外部からの書き込みによりデータの不整合が発生し、データ損失につながる可能性があります。

• 同期中に宛先の Kafka インスタンスをスケーリングした場合、タスクを再開するためにインスタンスを再起動してください。

• EncDB 機能が有効化された ApsaraDB RDS for MySQL インスタンスは、完全データ同期をサポートしていません。透過的データ暗号化 (TDE) が有効化されたインスタンスは、スキーマ同期、完全データ同期、および増分データ同期をサポートしています。

• DTS タスクが失敗した場合、DTS サポートは 8 時間以内に復旧を試みます。復旧中、タスクが再起動され、タスクパラメーターが変更される可能性があります。データベースパラメーターは変更されません。

• タスク実行中に自己管理 MySQL ソースでプライマリ/セカンダリ スイッチオーバーを実行すると、タスクは失敗します。

• DTS は、宛先データベースに同期された最新データのタイムスタンプと、ソースデータベースの現在のタイムスタンプに基づいて同期遅延を計算します。ソースデータベースで長時間 DML 操作が実行されない場合、同期遅延が正確でなくなる可能性があります。遅延が過大に表示される場合は、ソースデータベースで DML 操作を実行して遅延を更新できます。同期対象オブジェクトとしてデータベース全体を選択した場合は、ハートビートテーブルを作成することもできます。ハートビートテーブルは毎秒更新またはデータを受信します。

• DTS は、バイナリログファイルの位置を進めるために、ソースデータベースで定期的に CREATE DATABASE IF NOT EXISTS 'test' を実行します。これは想定された動作です。

レコードサイズの制限

Kafka に書き込まれる単一レコードの最大サイズは 10 MB です。ソースの行が 10 MB を超える場合、タスクは中断されます。これを回避するには、タスク構成時にフィルター条件を使用して大容量フィールドのカラムを除外してください。すでに大容量フィールドを含むテーブルがタスクオブジェクトに含まれている場合は、そのテーブルを削除して再度追加し、大容量フィールドを除外するフィルター条件を設定してください。

ステップ 1：データ同期ページへの移動

DTS コンソールまたは DMS コンソールのいずれかを使用します。

DTS コンソール

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

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

3. 左上隅で、同期インスタンスを配置するリージョンを選択します。

DMS コンソール

正確な手順は、DMS コンソールのモードおよびレイアウトによって異なる場合があります。「シンプルモード」および「DMS コンソールのレイアウトとスタイルのカスタマイズ」をご参照ください。

1. DMS コンソールにログインします。

2. 上部のナビゲーションバーで、Data + AI にポインターを合わせ、DTS (DTS) > データ同期 を選択します。

3. データ同期タスク の右側にあるドロップダウンリストから、同期インスタンスを配置するリージョンを選択します。

ステップ 2：ソースおよび宛先データベースの構成

1. タスクの作成 をクリックします。

2. 次の表のパラメーターを使用して、ソースおよび宛先データベースを構成します。

   警告

   ソースおよび宛先データベースを構成した後、タスクの失敗やデータの不整合を回避するために、ページに表示される 制限事項 を確認してください。

   セクション	パラメーター	説明
   該当なし	タスク名	DTS タスクの名前です。DTS は自動的に名前を生成します。タスクを識別しやすくするために、わかりやすい名前を指定してください。名前は一意である必要はありません。
   ソースデータベース	既存の接続を選択	インスタンスが DTS に登録されている場合は、ドロップダウンリストから選択すると、DTS が残りのパラメーターを自動的に入力します。そうでない場合は、データベースパラメーターを手動で構成してください。DMS コンソールでは、DMS データベースインスタンスの選択 ドロップダウンリストから選択します。
   	データベースタイプ	MySQL を選択します。
   	アクセス方法	Alibaba Cloud インスタンス を選択します。
   	インスタンスリージョン	ソース ApsaraDB RDS for MySQL インスタンスが配置されているリージョンです。
   	Alibaba Cloud アカウント間でデータを複製	同一アカウント内での同期の場合は、いいえ を選択します。
   	RDS インスタンス ID	ソース ApsaraDB RDS for MySQL インスタンスの ID です。
   	データベースアカウント	同期対象オブジェクトに対する読み取り権限を持つデータベースアカウントです。
   	データベースパスワード	データベースアカウントのパスワードです。
   	暗号化	[非暗号化] または [SSL 暗号化] を選択します。SSL 暗号化を使用するには、DTS タスクを設定する前に、RDS インスタンスで SSL を有効化してください。詳細については、「クラウド証明書を使用して SSL 暗号化を有効化する」をご参照ください。
   宛先データベース	既存の接続を選択	インスタンスが DTS に登録されている場合は、ドロップダウンリストから選択します。そうでない場合は、データベースパラメーターを手動で構成してください。
   	データベースタイプ	Kafka を選択します。
   	アクセス方法	Alibaba Cloud インスタンス を選択します。
   	インスタンスリージョン	宛先 ApsaraMQ for Kafka インスタンスが配置されているリージョンです。
   	Kafka インスタンス ID	宛先 ApsaraMQ for Kafka インスタンスの ID です。
   	暗号化	セキュリティ要件に基づいて、非暗号化 または SCRAM-SHA-256 を選択します。
   	トピック	同期データを受信するトピックです。ドロップダウンリストから選択します。
   	DDL 情報を格納するトピック	DDL 情報を格納するトピックです。空白のままにすると、DDL 情報は トピック で指定されたトピックに格納されます。
   	Kafka Schema Registry を使用	Avro スキーマの格納および取得に Kafka Schema Registry を使用するかどうかを指定します。いいえ または はい を選択します。はい を選択する場合、Avro スキーマ用に Kafka Schema Registry に登録された URL または IP アドレスを入力します。

3. 接続テストして次へ をクリックします。

   DTS サーバーの CIDR ブロックが、ソースデータベースとターゲットデータベースの両方のセキュリティ設定に追加されていることを確認してください。詳細については、「DTS サーバーの CIDR ブロックを追加する」をご参照ください。アクセス方法として Alibaba Cloud インスタンス を使用しない自己管理データベースの場合は、[接続テスト] を [DTS サーバーの CIDR ブロック] ダイアログボックスでクリックします。

ステップ 3：同期オブジェクトの構成

1. オブジェクトの構成 ステップで、同期パラメーターを設定します。

   パラメーター	説明
   同期タイプ	増分データ同期 がデフォルトで選択されています。フルデータ同期 も選択して、既存データを増分同期のベースラインとして同期してください。 説明 送信先が ApsaraMQ for Kafka インスタンスの場合、スキーマ同期は利用できません。
   競合テーブルの処理モード	事前チェックとエラー報告データベース、テーブル、およびカラム名のマッピング：送信先にソーステーブルと同じ名前のテーブルが存在する場合、事前チェックが失敗します。競合するテーブルの名前を変更するには、オブジェクト名マッピング機能を使用してください。詳細については、「」をご参照ください。エラーを無視して続行：重複テーブル名の事前チェックをスキップします。フル同期中は、競合するレコードが上書きされず、送信先の既存レコードが保持されます。増分同期中は、競合するレコードが送信先の既存レコードを上書きします。スキーマが異なる場合、同期が失敗するか、一部のカラムのみが同期される可能性があります。慎重に操作してください。
   Kafka のデータ形式	Kafka に書き込まれるメッセージ形式です。DTS Avro：DTS Avro スキーマ定義に従ってデータが構造化されます。スキーマの詳細は GitHub をご確認ください。Canal Json：データは Canal JSON 形式で保存されます。詳細については、「Canal Json」をご参照ください。
   Kafka データ圧縮形式	Kafka メッセージの圧縮アルゴリズムです。LZ4（デフォルト）：圧縮率は低く、高速です。GZIP：圧縮率は高く、低速で、CPU リソースを大量に消費します。Snappy：圧縮率と速度は中間レベルです。
   Kafka パーティションへのデータ送信ポリシー	Kafka パーティションへのデータ移行ポリシーの指定レコードを Kafka パーティションに分散する方法です。詳細については、「」をご参照ください。
   メッセージ確認応答メカニズム	Kafka プロデューサーの確認応答設定です。詳細については、「メッセージ確認応答メカニズム」をご参照ください。
   送信先インスタンスでのオブジェクト名の大文字・小文字処理	送信先のデータベース、テーブル、およびカラム名の大文字・小文字を制御します。デフォルトは DTS デフォルトポリシー宛先インスタンスにおけるオブジェクト名の英字大文字・小文字の指定 です。詳細については、「」をご参照ください。
   ソースオブジェクト	1 つ以上のオブジェクトを選択し、アイコンをクリックして 選択済みオブジェクト に追加します。同期対象オブジェクトとしてテーブルを選択できます。
   選択済みオブジェクト	選択されたオブジェクトの一覧です。各ソーステーブルに対して、送信先トピック名、パーティション数、およびパーティションキーを指定するには、オブジェクト名マッピング機能を使用してください。詳細については、「オブジェクト名マッピング機能の使用」をご参照ください。オブジェクトごとに SQL 操作をフィルターするには、選択済みオブジェクト セクション内のオブジェクトを右クリックし、同期する操作を選択します。

2. 次へ：詳細設定 をクリックし、詳細パラメーターを構成します。

   パラメーター	説明
   タスクスケジューリングの専用クラスター	デフォルトでは、DTS はタスクを共有クラスターにスケジュールします。安定性を向上させるには、専用クラスターを購入して選択してください。詳細については、「DTS 専用クラスターとは」をご参照ください。
   接続失敗時の再試行時間	タスク開始後に DTS が接続失敗を再試行する時間を指定します。有効値：10～1,440 分。デフォルト：720 分。少なくとも 30 分に設定してください。この期間内に DTS が再接続すると、タスクは再開されます。それ以外の場合、タスクは失敗します。複数のタスクで同じソースまたは宛先を共有する場合、最も短い再試行時間が適用されます。注：再試行期間中も DTS インスタンスに対して課金されます。ビジネス要件に基づいて再試行時間を指定し、ソースおよび宛先インスタンスをリリースした後は、速やかに DTS インスタンスをリリースすることを推奨します。
   その他の問題の再試行時間	DDL または DML 操作の失敗を DTS が再試行する時間を指定します。有効値：1～1,440 分。デフォルト：10 分。少なくとも 10 分に設定してください。この値は 接続失敗時の再試行時間 より小さくする必要があります。
   完全データ同期のスロットリングを有効化	完全同期中の読み取り/書き込みスループットを制限して、ソースおよび宛先サーバーの負荷を軽減します。ソースデータベースへのクエリ/秒 (QPS)、完全データ移行の RPS、および 完全移行のデータ移行速度 (MB/s) を構成します。完全データ同期 が選択されている場合にのみ利用可能です。
   増分データ同期のスロットリングを有効化	増分同期中のスループットを制限します。増分データ同期の RPS および 増分同期のデータ同期速度 (MB/s) を構成します。
   転送タスクと逆再生タスクのハートビートテーブルに対する SQL 操作を削除するかどうか	DTS がソースデータベースにハートビート SQL 操作を書き込むかどうかを制御します。はい：ハートビート操作を書き込みません。タスクに遅延インジケーターが表示される可能性があります。いいえ：ハートビート操作を書き込みます。ソースデータベースの物理バックアップおよびクローンに影響を与える可能性があります。
   環境タグ	DTS インスタンスを識別するためのオプションのタグです。
   ETL の構成	抽出・変換・書き出し（ETL）を有効にするかどうか。はい: データ処理の文を入力するためのコードエディタが開きます。詳細については、「データ移行またはデータ同期タスクで ETL を設定する」をご参照ください。いいえ: ETL は無効になります。
   モニタリングとアラート	タスクのアラートを構成するかどうかを指定します。はいDTS タスク作成時のモニタリングとアラートの設定：アラートのしきい値と通知先連絡先を設定します。「」をご参照ください。いいえ：アラートは無効になります。

ステップ 4：事前チェックの実行

1. 次へ：タスク設定の保存と事前チェック をクリックします。

   このタスク構成の API パラメーターをプレビューするには、次へ：タスク設定の保存と事前チェック にポインターを合わせ、OpenAPI パラメーターのプレビュー をクリックします。

2. 事前チェック結果を確認します。

   • すべての項目が合格した場合は、次のステップに進んでください。

   • 項目が不合格になった場合は、不合格項目の横にある 詳細を表示 をクリックして問題を解決し、再度事前チェック をクリックします。

   • 無視できない項目でアラートがトリガーされた場合は、問題を解決して事前チェックを再実行してください。無視可能なアラートの場合は、アラート項目の横にある 詳細を表示 をクリックし、次に 無視 > OK をクリックしてから、再度事前チェック をクリックします。

   警告

   事前チェックのアラートを無視すると、データの不整合が発生する可能性があります。

ステップ 5：インスタンスの購入とタスクの開始

1. 成功率 が 100% に達したら、次へ：インスタンスの購入 をクリックします。

2. 購入 ページで、課金およびインスタンスパラメーターを構成します。

   パラメーター	説明
   Billing Method	サブスクリプション：固定期間分を前払いします。長期利用の場合、コスト効率に優れています。サブスクリプション期間のオプション：1～9か月、または1年、2年、3年、5年。従量課金：1時間単位で課金されます。短期利用に適しています。不要になった時点でインスタンスをリリースすると、課金が停止されます。
   Resource Group Settings	同期インスタンスのリソースグループです。デフォルト：default resource group。詳細については、「What is Resource Management?」をご参照ください。
   Instance Class	インスタンスクラスによって同期速度が異なります。詳細については、「Instance classes of data synchronization instances」をご参照ください。

3. 読み、[Data Transmission Service (従量課金) サービス利用規約] を選択してください。

4. 購入して開始 をクリックし、ダイアログボックスで OK をクリックします。

タスクがタスクリストに表示されます。そこから進行状況を追跡できます。