すべてのプロダクト
Search

このプロダクト

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

    ドキュメントセンター

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

    最終更新日:Mar 29, 2026

    Data Transmission Service (DTS) を使用して、ApsaraDB RDS for MySQL インスタンスからの変更データを ApsaraMQ for Kafka インスタンスにストリーミングします。DTS はバイナリログを介して行レベルの変更をキャプチャし、リアルタイムで Kafka トピックに配信することで、ダウンストリーム分析、イベント駆動型アーキテクチャ、およびデータパイプラインの統合を可能にします。

    前提条件

    開始する前に、以下を確認してください。

    • ApsaraDB RDS for MySQL インスタンス。設定手順については、「ApsaraDB RDS for MySQL インスタンスの作成」をご参照ください。

    • 同期されたデータを受信するためにトピックが作成された ApsaraMQ for Kafka インスタンス。「ステップ 1：トピックの作成」をご参照ください。サポートされているソースとターゲットのバージョンについては、「データ同期シナリオの概要」をご参照ください。

    • Kafka インスタンスに十分な空きストレージ容量があること。少なくとも RDS MySQL インスタンスの総データサイズと同じ大きさが必要です。

    • ソースインスタンスでバイナリログ記録が有効になっており、binlog_row_image が full に設定されています。ApsaraDB RDS for MySQL では、バイナリログ記録がデフォルトで有効になっています。このパラメーターの確認または変更については、「インスタンスパラメーターの変更」をご参照ください。

    制限事項

    ソースデータベース

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

    • データベース全体ではなく個別のテーブルを選択し、同期中にテーブル名またはカラム名を変更する場合、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; を実行します。明示的なプライマリキーがないテーブルは、不可視のプライマリキーを自動生成する場合があります。それらも可視にしてください。「Invisible Columns」および「Generated Invisible Primary Keys」をご参照ください。

    • 読み取り専用の 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 インスタンスは、完全データ同期をサポートしていません。Transparent Data Encryption (TDE) が有効になっているインスタンスは、スキーマ同期、完全データ同期、および増分データ同期をサポートしています。

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

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

    • DTS は、ターゲットデータベースの最新の同期済みデータのタイムスタンプとソースデータベースの現在のタイムスタンプに基づいて同期遅延を計算します。ソースデータベースで長期間 DML 操作が実行されない場合、同期遅延が不正確になることがあります。遅延が大きすぎるように見える場合は、ソースデータベースで DML 操作を実行して遅延を更新できます。データベース全体を同期オブジェクトとして選択した場合は、ハートビートテーブルを作成することもできます。ハートビートテーブルは毎秒更新されるか、データを受信します。

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

    レコードサイズの制限

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

    課金

    同期タイプ料金
    スキーマ同期と完全データ同期無料
    増分データ同期課金済み。詳細については、「課金概要」をご参照ください。

    サポートされる同期トポロジ

    • 一方向 1 対 1 同期

    • 一方向 1 対多同期

    • 一方向多対 1 同期

    サポートされているすべてのトポロジについては、「同期トポロジ」をご参照ください。

    同期可能な SQL 操作

    操作タイプSQL 文
    DMLINSERT、UPDATE、DELETE
    DDLCREATE TABLE、ALTER TABLE、DROP TABLE、RENAME TABLE、TRUNCATE TABLE; CREATE VIEW、ALTER VIEW、DROP VIEW; CREATE PROCEDURE、ALTER PROCEDURE、DROP PROCEDURE; CREATE FUNCTION、DROP FUNCTION、CREATE TRIGGER、DROP TRIGGER; CREATE INDEX、DROP INDEX
    DTS はソースからターゲットへ外部キーを同期しません。ソースでのカスケード操作と削除操作は複製されません。

    同期タスクの作成

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

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

    DTS コンソール

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

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

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

    DMS コンソール

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

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

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

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

    ステップ 2：ソースデータベースとターゲットデータベースの設定

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

    2. 次の表のパラメーターを使用して、ソースデータベースとターゲットデータベースを設定します。

      警告

      ソースデータベースとターゲットデータベースを設定した後、タスクの失敗やデータの不整合を避けるために、ページに表示される [使用制限] を確認してください。

      セクションパラメーター説明
      N/Aタスク名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 インスタンスをリリースすることをお勧めします。
      その他の問題に対する再試行時間DTS が失敗した DDL または DML 操作を再試行する時間。有効な値：1〜1,440 分。デフォルト：10 分。少なくとも 10 分に設定してください。この値は [接続失敗時の再試行時間] より小さくする必要があります。
      完全データ同期のスロットリングを有効にする完全同期中に読み取り/書き込みスループットを制限して、ソースサーバーおよび送信先サーバーの負荷を軽減します。[ソースデータベースへの QPS (クエリ/秒)]、[完全データ移行の RPS]、および[完全移行時のデータ移行速度 (MB/秒)] を設定します。完全データ同期が選択されている場合にのみ利用可能です。
      増分データ同期のスロットリングを有効にする増分同期中のスループットを制限します。[増分データ同期の RPS] および [増分同期のデータ同期速度 (MB/s)] を設定します。
      順方向および逆方向タスクのハートビートテーブルに対する SQL 操作を削除するかどうかDTS がハートビート SQL 操作をソースデータベースに書き込むかどうかを制御します。[はい]：ハートビート操作を書き込みません — タスクに遅延インジケーターが表示される場合があります。[いいえ]：ハートビート操作を書き込みます — ソースデータベースの物理バックアップとクローニングに影響を与える可能性があります。
      環境タグDTS インスタンスを識別するためのオプションのタグ。
      ETL の設定抽出・変換・書き出し (ETL) を有効にするかどうか。はい: データ処理文を入力するためのコードエディタが開きます。「データ移行またはデータ同期タスクで ETL を設定する」をご参照ください。いいえ: ETL は無効になります。
      モニタリングとアラートタスクのアラートを設定するかどうか。[はい]DTS タスク作成時のモニタリングとアラートの設定：アラートのしきい値と通知連絡先を設定します — 「」をご参照ください。[いいえ]：アラートは無効になります。

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

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

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

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

      • すべての項目が合格した場合、次のステップに進みます。

      • 項目が失敗した場合は、失敗した項目の横にある [詳細の表示] をクリックし、問題を解決してから [再試行] をクリックします。

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

      警告

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

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

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

    2. [購入] ページで、課金とインスタンスのパラメーターを設定します。

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

    3. [Data Transmission Service (従量課金) サービス規約] を読み、選択します。

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

    タスクがタスクリストに表示されます。そこから進捗を追跡します。

    オブジェクト名マッピング機能の使用

    この機能を使用して、ソーステーブルのデータを特定の Kafka トピックにルーティングし、パーティション数を制御し、パーティションキーを設定します。

    1. [選択したオブジェクト] セクションで、トピック名にポインターを合わせます。

    2. 右クリックして [編集] を選択します。

    3. [テーブルの編集] ダイアログボックスで、パラメーターを設定します。

      パラメーター説明
      テーブル名このソーステーブルからデータを受信するトピック。デフォルトでは、[ターゲットデータベース] セクションで設定されたトピックになります。ApsaraMQ for Kafka ターゲットの場合、トピックはすでに存在している必要があります — DTS は自動的に作成しません。この値を変更すると、ソーステーブルのデータが指定されたトピックにルーティングされます。
      フィルター条件このテーブルに対する SQL ベースの行フィルター。「フィルター条件の指定」をご参照ください。
      パーティション数ターゲットトピックのパーティション数。
      パーティションキーパーティションのハッシュ値を計算するために使用される 1 つ以上のカラム。[Kafka パーティションへのデータ転送ポリシー] が [プライマリキーのハッシュ値に基づいてデータを別々のパーティションに転送] に設定されている場合に適用されます。パーティションキーを設定するには、まず [すべてのテーブルを同期] の選択を解除します。

    4. [OK] をクリックします。

    よくある質問

    タスク開始後に Kafka データ圧縮形式を変更できますか？

    はい。「同期オブジェクトの変更」をご参照ください。

    タスク開始後にメッセージ確認応答メカニズムを変更できますか？

    はい。「同期オブジェクトの変更」をご参照ください。