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

ソースデータベース

増分移行のためのバイナリログ要件

増分データ移行を有効にする前に、これらの MySQL 設定を確認および構成してください：

セルフマネージド MySQL インスタンスの場合、対応するコマンドを実行して非準拠の設定を修正できます：

-- binlog 形式を設定
SET GLOBAL binlog_format = 'ROW';

-- binlog row image を設定
SET GLOBAL binlog_row_image = 'FULL';

その他の制限事項

• 移行するオブジェクトとしてテーブルのみを選択できます。

• ソースおよび送信先サーバーへの負荷を軽減するため、オフピーク時に移行を実行してください。

• 本番ワークロードを送信先に切り替える前に、過去 7 日間に失敗した DTS タスクを停止またはリリースしてください。DTS は失敗したタスクを自動的に再開しようとし、送信先のデータを上書きする可能性があります。または、送信先に対する DTS アカウントの書き込み権限を取り消してください。

• 移行中に他のソースから送信先にデータが書き込まれると、データの不整合が発生します。

• 移行中に送信先の Kafka クラスターがアップグレードまたはダウングレードされた場合は、移行タスクを再起動してください。

• ソースの RDS for MySQL インスタンスで TDE (透過的データ暗号化) が有効になっている場合、スキーマ移行、完全データ移行、増分データ移行はすべてサポートされます。代わりに EncDB 機能が有効になっている場合、完全データ移行はサポートされません。

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

特殊なケース

セルフマネージド MySQL：

• タスク実行中にプライマリ/セカンダリ スイッチオーバーが発生した場合、タスクは失敗します。

• ソースで長期間 DML 操作が実行されない場合、移行遅延が不正確に表示されることがあります。ソースで任意の DML 操作を実行して遅延を更新してください。データベース全体を移行した場合は、毎秒更新を受け取るハートビートテーブルを作成してください。

• DTS は、バイナリログの位置を進めるために、ソースで定期的に CREATE DATABASE IF NOT EXISTS \`test\` を実行します。

ApsaraDB RDS for MySQL：

• 読み取り専用の RDS for MySQL V5.6 インスタンス (トランザクションログを記録しない) は、増分データ移行のソースとして使用できません。

• DTS は、バイナリログの位置を進めるために、ソースで定期的に CREATE DATABASE IF NOT EXISTS \`test\` を実行します。

DTS Avro

レコードは DTS Avro スキーマ定義に従います。GitHub でスキーマをご参照ください。

Canal JSON

レコードは Canal JSON 形式で書き込まれます。次の例は、典型的な INSERT イベントを示しています：

{
  "data": [
    {
      "id": "1",
      "name": "example_value",
      "updated_at": "2026-01-01 00:00:00"
    }
  ],
  "database": "example_db",
  "es": 1735689600000,
  "id": 1,
  "isDdl": false,
  "mysqlType": {
    "id": "int(11)",
    "name": "varchar(255)",
    "updated_at": "datetime"
  },
  "old": null,
  "pkNames": ["id"],
  "sql": "",
  "sqlType": {
    "id": 4,
    "name": 12,
    "updated_at": 93
  },
  "table": "example_table",
  "ts": 1735689600123,
  "type": "INSERT"
}

UPDATE イベントの場合、old フィールドには変更前の値 (変更前の列の値) が含まれます。

完全なパラメーターの詳細とサンプルコンシューマーコードについては、「Canal JSON データ形式」をご参照ください。

ステップ 1：データ移行ページに移動

いずれかのコンソールを使用します：

ステップ 2：ソースデータベースとターゲットデータベースの構成

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

   警告

   ソースと送信先を構成した後、続行する前にページの上部に表示される [使用制限] をお読みください。このステップをスキップすると、タスクが失敗したり、データの不整合が発生したりする可能性があります。

2. 次の表のパラメーターを構成します。

ソースデータベース

宛先データベース

1. [接続をテストして続行] をクリックします。

   DTS はソースおよび送信先にアクセスできる必要があります。両方のデータベースのセキュリティ設定に DTS サーバーの CIDR ブロックを追加します。詳細については、「DTS サーバーの CIDR ブロックを追加する」をご参照ください。ソースまたは送信先が Alibaba Cloud インスタンス以外のアクセス方法を使用する自己管理データベースである場合は、[接続テスト] を [DTS サーバーの CIDR ブロック] ダイアログボックスでクリックします。

ステップ 3：移行オブジェクトの選択と設定の構成

1. [オブジェクトの構成] ページで、次のパラメーターを設定します。

1. [次へ：詳細設定] をクリックします。

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

1. このタスクの API パラメーターをプレビューするには、[次へ：タスク設定を保存して事前チェック] にカーソルを合わせ、[OpenAPI パラメーターのプレビュー] をクリックします。

2. [次へ：タスク設定を保存して事前チェック] をクリックします。DTS はタスクを開始する前に自動事前チェックを実行します。いずれかの項目が失敗した場合：

   • 失敗した項目の横にある [詳細を表示] をクリックし、問題を修正してから [再度事前チェック] をクリックします。

   • アラート項目が重大でなく、無視できる場合は、[アラート詳細の確認] > [無視] > [OK] をクリックし、その後 [再度事前チェック] をクリックします。アラートを無視すると、データの不整合につながる可能性があります。

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

1. [成功率] が 100% に達するのを待ってから、[次へ：インスタンスの購入] をクリックします。

2. [インスタンスの購入] ページで、以下を構成します：

1. [Data Transmission Service (従量課金) サービス規約] を読み、チェックボックスを選択して同意します。

2. [購入して開始] をクリックし、確認ダイアログで [OK] をクリックします。