Data Transmission Service (DTS) は、ApsaraDB for MongoDB レプリカセットインスタンスから AnalyticDB for MySQL 3.0 クラスターへデータを継続的に同期し、分析用データベースを最小限の遅延で最新状態に保ちます。
MongoDB からリレーショナルデータベースへの概念対応
MongoDB に慣れていない場合、本ガイド全体で以下の概念対応が適用されます:
| MongoDB の概念 | リレーショナルデータベースでの対応 | 備考 |
|---|---|---|
| コレクション | テーブル | 同期対象となるデータ単位 |
| ドキュメント | 行 | 各ドキュメントは、送信先テーブルの 1 行に対応します |
| フィールド | 列 | フィールドは bson_value() 式を使用してマッピングする必要があります |
| ObjectId | VARCHAR | _id フィールド(ObjectId 型)は VARCHAR 列にマッピングする必要があります |
DTS がサポートする機能
| 機能 | 対応済み | 備考 |
|---|---|---|
| 完全データ同期 | はい | 無料 |
| 増分データ同期 | はい | 課金対象。詳細については、「課金概要 |
| コレクション | はい | — |
| INSERT、UPDATE、DELETE 操作 | はい | 増分同期のみ対応。ファイルの増分データを同期する際は、$set コマンドのみ同期実行可能です |
| スキーマ同期 | いいえ | 開始前に送信先テーブルを手動で作成してください |
| インデックス、ビュー、プロシージャ、関数、トリガー、外部キー | いいえ | — |
| admin、config、local データベース | いいえ | — |
| トランザクション | いいえ | トランザクションは個別のレコードに変換されます |
前提条件
開始する前に、以下の点を確認してください。
シャードクラスターをソースとする場合: 各シャードノードにパブリックエンドポイントが設定されており、すべてのシャードアカウントで同一のユーザー名とパスワードが使用されていること。詳細については、「シャードクラスターインスタンスにおけるシャードノードまたは ConfigServer ノードのエンドポイント申請」をご参照ください。
送信先の AnalyticDB for MySQL 3.0 クラスターが存在し、その利用可能なストレージ容量がソースの ApsaraDB for MongoDB インスタンスの使用済みストレージ容量より大きいこと。(推奨:ソースデータの合計サイズより少なくとも 10 % 多い容量)詳細については、「クラスターの作成」をご参照ください。
送信先のデータベースおよび、一意(複合でない)なプライマリキー列を持つテーブルが存在すること。詳細については、「CREATE DATABASE」および「CREATE TABLE」をご参照ください。
送信先クラスターで複数ステートメント機能が有効化されていること:
SET ADB_CONFIG ALLOW_MULTI_QUERIES=true;
複数ステートメント機能を使用するには、AnalyticDB for MySQL V3.1.9.3 以降が必要です。詳細については、「クラスターのマイナーバージョンを更新する」をご参照ください。
送信先テーブルを作成する場合:
ObjectId 型の
_idフィールドを VARCHAR 列にマッピングします。任意の列名を
_idまたは_valueとしないでください。すべての送信先カラムのデータの型が、ソース MongoDB のデータの型と互換性があることを確認してください。
データ型マッピング
送信先テーブルを作成する前に、各 AnalyticDB for MySQL カラムの型がソース MongoDB のデータの型と互換であることを確認してください。
| MongoDB BSON 型 | AnalyticDB for MySQL 型 | 備考 |
|---|---|---|
| ObjectId | VARCHAR | _id プライマリキー列。ObjectId 値は文字列として格納されます |
| FLOAT | FLOAT | DTS ではデフォルトで ROUND(column, 38) を使用します。必要に応じて列の精度を調整してください |
| DOUBLE | DOUBLE | DTS ではデフォルトで ROUND(column, 308) を使用します。必要に応じて列の精度を調整してください |
| ネストされたオブジェクトフィールド | 個別の列 | 各サブフィールドは bson_value("parent", "child") を使用してマッピングします — 親オブジェクトを直接マッピングしないでください |
サポートされるデータ型マッピングの完全な一覧については、「ApsaraDB for MongoDB インスタンスから ApsaraDB RDS for MySQL インスタンスへのデータ移行」の「データ型マッピング」セクションをご参照ください。
必要な権限
| データベース | 必要な権限 | リファレンス |
|---|---|---|
| ソースの ApsaraDB for MongoDB インスタンス | ソース、admin、local データベースに対する読み取り権限 | アカウント管理 |
| 送信先の AnalyticDB for MySQL 3.0 クラスター | 送信先データベースに対する読み取りおよび書き込み権限 | データベースアカウントの作成 |
制限事項
ソースデータベースの制限
帯域幅: ソースデータベースをホストするサーバーには十分なアウトバウンド帯域幅が必要です。帯域幅が低いと同期速度が低下します。
コレクション数の制限: 送信先コレクションの名前マッピングを構成した場合、1 つのタスクで最大 1,000 個のコレクションを同期できます。それ以上のコレクションを同期する場合は、複数のタスクを作成するか、タスクをデータベース全体の同期に構成してください。
シャード化クラスター — `_id` の一意性: 同期対象となる各コレクションの
_idフィールドは一意である必要があります。_id値が重複すると、データの不整合が発生します。シャードクラスター — Mongos ノード: Mongos ノードの数は 10 を超えてはなりません。ソースインスタンスには孤立ドキュメントを含めてはいけません。そうしないと、データの不整合が発生し、タスクが失敗する可能性があります。詳細については、「MongoDB ドキュメント」および「孤立ドキュメントの削除に関する FAQ」をご参照ください。
シャードクラスター — バランサーのレイテンシー: シャードクラスターのソースでバランサーがデータのバランス調整を行っている場合、インスタンスにレイテンシーが発生する可能性があります。
非対応のソース: スタンドアロンの ApsaraDB for MongoDB インスタンス、Azure Cosmos DB for MongoDB クラスター、Amazon DocumentDB エラスティッククラスターはソースとして使用できません。
SRV エンドポイント: DTS は SRV 接続文字列経由で MongoDB データベースに接続できません。
oplog / change streams: ソースデータベースには oplog が有効化されており、少なくとも 7 日分のログデータが保持されている必要があります。あるいは、change streams が有効化され、過去 7 日間のデータをカバーしている必要があります。
可能であれば、ソースデータの変更を記録するために oplog を使用してください。change streams は MongoDB 4.0 以降で必要であり、双方向同期はサポートしていません。非エラスティックの Amazon DocumentDB クラスターでは、change streams を使用する必要があり、移行方法 を ChangeStream に、アーキテクチャ を Sharded Cluster に設定する必要があります。
同期中の制限:
完全データ同期中は、コレクションのスキーマを変更したり、ARRAY 型のデータを変更したりしないでください。
完全データ同期のみ(増分同期なし)を実行している場合、タスク実行中にソースデータベースへの書き込みを行わないでください。
シャードクラスターをソースとする場合、タスク実行中に
shardCollection、reshardCollection、unshardCollection、moveCollection、movePrimaryを実行しないでください。
その他の制限
送信先テーブルのプライマリキーには、選択されたオブジェクト の構成で
bson_value("_id")を割り当てる必要があります。_idまたは_valueという名前の送信先列は、同期失敗の原因となります。
AnalyticDB for MySQL 3.0 の任意のノードにおけるディスク使用率が 80 % を超えると、DTS タスクは遅延し、エラーを返します。タスク開始前に必要なディスク領域を見積もってください。
送信先クラスターのバックアップ実行中に DTS タスクを実行すると、タスクは失敗します。
完全データ同期中の同時 INSERT 操作により、コレクションの断片化が発生します。送信先データベースのストレージ使用量は、ソースよりも大きくなります。
FLOAT および DOUBLE の精度:DTS では
ROUND(COLUMN, PRECISION)を使用します。FLOAT のデフォルト精度は 38 桁、DOUBLE のデフォルト精度は 308 桁です。ビジネス要件に応じて、列構成で精度を指定してください。トランザクションは個別のレコードに変換されます。トランザクションコンテキストは保持されません。
増分同期遅延:DTS は、最新の同期済みデータタイムスタンプと現在のソースタイムスタンプとの差に基づいて遅延を計算します。ソースに長期間更新がない場合、遅延の表示値は不正確になる可能性があります。遅延の表示値を更新するには、ソースに対して更新操作を実行してください。
DTS は、失敗したタスクを最大 7 日間再開しようと試みます。ワークロードを送信先データベースに切り替える前に、失敗したタスクを停止または解放するか、DTS アカウントの書き込み権限を REVOKE で削除して、ソースデータによる送信先データの上書きを防いでください。
DTS テクニカルサポートは、失敗したタスクを 8 時間以内に復元しようと試みます。復元中、タスクが再起動される場合や、タスクパラメーター(データベースパラメーターではありません)が変更される場合があります。詳細については、「インスタンスパラメーターの変更」をご参照ください。
ソースおよび送信先データベースのパフォーマンスへの影響を軽減するため、非ピーク時間帯にデータ同期を実行してください。
同期タスクの設定
ステップ 1:データ同期ページを開く
タスク設定ページに移動するには、以下のいずれかの方法を使用します。
DTS コンソール
DMS コンソール
利用可能なオプションは、DMS コンソールのモードとレイアウトによって異なります。「シンプルモード」と「DMS コンソールのレイアウトおよびスタイルのカスタマイズ」を参照してください。
ステップ 2:ソースおよび送信先データベースの構成
タスクの作成 をクリックします。
以下の表のパラメーターを使用して、タスク名およびデータベース接続を構成します。
| パラメーター | 説明 |
|---|---|
| タスク名 | わかりやすい名前を入力します。DTS では自動的に名前が生成されますが、意味のある名前を付けることでタスクの識別が容易になります。名前は一意である必要はありません。 |
ソースデータベース
| パラメーター | 説明 |
|---|---|
| 既存の接続の選択 | 登録済みの DTS データベースインスタンスを選択して、接続パラメーターを自動入力します。インスタンスが登録されていない場合は、パラメーターを手動で入力します。DMS コンソールでは、DMS データベースインスタンスの選択 から選択します。 |
| データベースタイプ | MongoDB を選択します。 |
| アクセス方法 | Alibaba Cloud インスタンス を選択します。 |
| インスタンスリージョン | ソースの ApsaraDB for MongoDB インスタンスのリージョンを選択します。 |
| Alibaba Cloud アカウント間でのデータ複製 | ソースデータベースが同一の Alibaba Cloud アカウント内にある場合は、いいえ を選択します。 |
| アーキテクチャ | 本例では レプリカセット を選択します。ソースがシャードクラスターを使用している場合は、シャードクラスター を選択し、シャードアカウント および シャードパスワード のパラメーターを構成します。 |
| 移行方法 | 増分データ同期に使用する方法を選択します:Oplog(oplog が有効な場合に推奨)または ChangeStream(change streams が有効な場合)。詳細については、「Change Streams」をご参照ください。 重要 シャードクラスター を アーキテクチャ として選択し、ChangeStream を 移行方法 として選択した場合、シャードアカウント および シャードパスワード のパラメーターは不要です。シャードクラスター を Oplog とともに選択した場合、SSL 暗号化接続は利用できません。 |
| インスタンス ID | ソースの ApsaraDB for MongoDB インスタンスの ID を選択します。 |
| 認証データベース | 認証データベースの名前を入力します。アカウントが変更されていない場合は、admin を入力します。 |
| データベースアカウント | データベースアカウントを入力します。「必要な権限」をご参照ください。 |
| データベースパスワード | アカウントのパスワードを入力します。 |
| 暗号化 | 要件に応じて、暗号化なし、SSL 暗号化、または Mongo Atlas SSL を選択します。利用可能なオプションは、アクセス方法 および アーキテクチャ の値によって異なります。ソースが自己管理デプロイメントのレプリカセットアーキテクチャを使用しており、SSL 暗号化 を選択した場合、接続を検証するために認証局 (CA) 証明書をアップロードできます。 |
送信先データベース
| パラメーター | 説明 |
|---|---|
| 既存の接続の選択 | 登録済みの DTS データベースインスタンスを選択して、接続パラメーターを自動入力します。インスタンスが登録されていない場合は、パラメーターを手動で入力します。 |
| データベースタイプ | AnalyticDB for MySQL 3.0 を選択します。 |
| アクセス方法 | Alibaba Cloud インスタンス を選択します。 |
| インスタンスリージョン | 送信先の AnalyticDB for MySQL 3.0 クラスターのリージョンを選択します。 |
| インスタンス ID | 送信先の AnalyticDB for MySQL 3.0 クラスターの ID を選択します。 |
| データベースアカウント | データベースアカウントを入力します。「必要な権限」をご参照ください。 |
| データベースパスワード | アカウントのパスワードを入力します。 |
接続性のテストと続行 をクリックします。
接続性テストが成功する前に、DTS サーバーの CIDR ブロックをソースデータベースおよびターゲットデータベースのセキュリティ設定に追加する必要があります。詳細については、「DTS サーバーの CIDR ブロックを追加する」をご参照ください。自己管理データベースの場合は、[接続性テスト] を [DTS サーバーの CIDR ブロック] ダイアログボックスでクリックします。
ステップ 3:同期対象オブジェクトの構成
オブジェクトの構成 ステップで、以下のパラメーターを構成します。
| パラメーター | 説明 |
|---|---|
| 同期タイプ | 増分データ同期 がデフォルトで選択されています。必要に応じて、完全データ同期 も選択できます。スキーマ同期は利用できません — 送信先テーブルは事前に手動で作成してください。 |
| 同期対象の DDL および DML 操作 | 増分データ同期中に同期する DDL および DML 操作を選択します。コレクションごとに操作を構成するには、選択されたオブジェクト 内のコレクションを右クリックし、ダイアログボックスで操作を選択します。 |
| テーブルのマージ | 複数のソーステーブルを同じスキーマで 1 つの送信先テーブルにマージする場合は、はい を選択します(OLAP シナリオで有用)。ソースデータベースから複数のテーブルを選択した後は、オブジェクト名マッピング機能を使用して、これらのテーブルの名前を送信先テーブルの名前に変更する必要があります。「オブジェクト名のマッピング」をご参照ください。DTS は、各ソースを識別するために __dts_data_source 列(TEXT)を追加し、<DTS インスタンス ID>:<ソース DB>:<ソーススキーマ>:<ソーステーブル> の形式で値を格納します(例:dts********:dtstestdata:testschema:customer1)。特定のソーステーブルをマージしない場合は、それらのテーブル用に個別のデータ同期タスクを作成してください。いいえ(デフォルト)を選択すると、各ソーステーブルが個別に同期されます。警告 マージが有効な状態で、ソーステーブルのスキーマを変更する DDL 操作を実行しないでください。これにより、データの不整合やタスクの失敗が発生する可能性があります。 |
| 競合テーブルの処理モード | 事前チェックとエラー報告(デフォルト):送信先にソーステーブルと同じ名前のテーブルが存在する場合、事前チェックが失敗します。エラーを無視して続行:このチェックをスキップします。 警告 エラーを無視して続行 を選択すると、データの不整合が発生する可能性があります。完全同期中は、送信先の競合する行が保持されます。増分同期中は、それらの行が上書きされます。スキーマが異なる場合、タスクが失敗するか、一部の列のみが同期される可能性があります。 |
同期対象オブジェクトの選択
ソースオブジェクト で、1 つ以上のコレクションを選択し、
をクリックして 選択されたオブジェクト に移動します。コレクションのみを選択できます。スキーマ名を編集します。
選択されたオブジェクト で、コレクションを含むデータベースを右クリックします。

スキーマ名 を、AnalyticDB for MySQL 3.0 クラスター内の送信先データベースの名前に設定します。

(任意)同期対象の DDL および DML 操作の選択 で、同期する操作を選択します。

OK をクリックします。
テーブル名を編集します。
選択されたオブジェクト 内のコレクションを右クリックします。

テーブル名 を、AnalyticDB for MySQL 3.0 クラスター内の送信先テーブルの名前に設定します。

(任意)完全データ同期のフィルター条件を構成します。「フィルター条件の設定」をご参照ください。

(任意)同期対象の DDL および DML 操作の選択 で、同期する操作を選択します。

ソースフィールドを送信先列にマッピングします。DTS はコレクションフィールドに対して自動的に
bson_value()式を生成します。各列の 列名、型、長さ、精度 の値を確認・調整してください。同期不要なフィールドを削除するには、該当行の横にある
アイコンをクリックします。自動生成された式が要求されるフィールド(例:ネストされたフィールド)と一致しない場合は、
アイコンをクリックし、+ 列の追加 をクリックします。
列名、型、長さ、精度 を構成します。
値の割り当て に
bson_value()式を入力します。「ネストされたフィールドのマッピング」で式の構文をご確認ください。追加の列について同様の手順を繰り返します。
重要送信先テーブルのプライマリキー列には
bson_value("_id")を割り当てます。各bson_value()式では、完全なフィールド階層を指定してください — 親フィールドのみ(例:bson_value("person"))を指定すると、増分サブフィールドデータが失われる可能性があります。OK をクリックします。
ステップ 4:高度な設定の構成
次へ:高度な設定 をクリックし、以下のパラメーターを構成します。
| パラメーター | 説明 |
|---|---|
| タスクスケジューリング用専用クラスター | デフォルトでは、DTS は共有クラスター上でタスクをスケジュールします。同期の安定性を向上させるには、専用クラスターをご購入ください。詳細については、「DTS 専用クラスターとは」をご参照ください。 |
| 接続失敗時のリトライ時間 | タスク開始後に DTS が接続失敗をリトライする時間です。有効値:10~1440 分。デフォルト:720 分。30 分以上に設定してください。この時間枠内で DTS が再接続した場合、タスクは再開されます。 説明 複数のタスクでソースデータベースまたはターゲットデータベースを共有している場合、最も短いリトライ時間枠が適用されます。リトライ期間中も DTS の課金は継続されます。 |
| その他の問題発生時のリトライ時間 | DTS が失敗した DDL または DML 操作をリトライする時間です。有効値:1~1440 分。デフォルト:10 分。10 分以上に設定してください。この値は 接続失敗時のリトライ時間 より小さくする必要があります。 |
| 同期対象データのテーブルにおけるプライマリキー `_id` のデータの型を 1 種類に限定する | コレクション内のすべてのドキュメントにおいて、_id プライマリキーのデータの型が一様であるかどうかを指定します。はい:DTS はデータの型のスキャンをスキップし、コレクションごとに 1 種類のデータの型のみを同期します。いいえ:DTS は _id のデータの型をスキャンし、すべてのドキュメントを同期します。このパラメーターは 完全データ同期 を選択した場合にのみ表示されます。正確に設定してください。誤った値を設定すると、データ損失が発生する可能性があります。 |
| 完全データ同期時のスロットリングを有効化 | 完全同期中にソースデータベースおよびターゲットデータベースへの負荷を制限するために、ソースデータベースへのクエリ数 (QPS)、完全データ移行の RPS、および 完全移行時のデータ移行速度 (MB/s) を設定します。このオプションは 完全データ同期 を選択した場合にのみ利用可能です。 |
| 増分データ同期時のスロットリングを有効化 | 増分同期中に負荷を制限するために、増分データ同期の RPS および 増分同期時のデータ同期速度 (MB/s) を設定します。 |
| 環境タグ | DTS インスタンスの環境を識別するための任意のタグです。 |
| ETL の設定 | はい を選択して ETL (抽出・変換・書き出し) 機能を有効にし、データ処理文を入力します。詳細については、「データ移行またはデータ同期タスクでの ETL の設定」をご参照ください。いいえ を選択すると、この機能をスキップします。 |
| モニタリングとアラート | はい を選択すると、タスクが失敗した場合や同期遅延がしきい値を超えた場合に通知を受け取れます。アラートのしきい値および 通知設定 を構成してください。詳細については、「DTS タスク作成時のモニタリングとアラートの設定」をご参照ください。 |
ステップ 5:設定の保存と事前チェックの実行
このタスク構成の API パラメーターをプレビューするには、次へ:タスク設定の保存と事前チェック の上にポインターを合わせ、OpenAPI パラメーターのプレビュー をクリックします。
次へ:タスク設定の保存と事前チェック をクリックして設定を保存し、事前チェックを開始します。
事前チェックに合格するまで、タスクは開始できません。
事前チェックが失敗した場合は、各失敗項目の横にある 詳細の表示 をクリックし、問題を解決してから再度事前チェックを実行してください。
事前チェック項目にアラートが表示された場合:
アラートを無視できない場合は、問題を解決してから再度事前チェックを実行してください。
アラートを無視できる場合は、アラートの詳細の確認 をクリックし、ダイアログボックスで 無視 をクリックし、OK をクリックしてから、再び事前チェック をクリックしてください。
ステップ 6:インスタンスの購入と開始
成功率 が 100 % に達したら、次へ:インスタンスの購入 をクリックします。
購入 ページで、課金方法およびインスタンスクラスを構成します。
| パラメーター | 説明 |
|---|---|
| 課金方法 | サブスクリプション:固定期間分の料金を前払いする方式。長期利用に適しており、コスト効率が優れます。従量課金:1 時間単位で課金される方式。短期利用に適しています。不要になった場合は、インスタンスをリリースして課金を停止してください。 |
| リソースグループ設定項目 | 同期インスタンスが属するリソースグループです。デフォルト値: デフォルトリソースグループ。詳細については、「Resource Management とは?」をご参照ください。 |
| インスタンスクラス | データ同期インスタンスのインスタンスタイプ必要な同期速度に応じてインスタンスクラスを選択します。詳細については、「」をご参照ください。 |
| サブスクリプション期間 | サブスクリプション課金方法でのみ利用可能です。有効な値:1~9 ヶ月、または 1 年、2 年、3 年、5 年。 |
Data Transmission Service(従量課金)サービス利用規約 を読み、同意します。
購入して開始 をクリックし、確認ダイアログボックスで OK をクリックします。
タスクはタスクリストに表示されます。そこで進行状況を監視できます。
ネストされたフィールドのマッピング
自動生成された bson_value() 式がソースデータ構造と一致しない場合、このリファレンスをご活用ください。
すべての bson_value() 式は、最上位のフィールドからサブフィールドまで、各レベルを個別の引数として指定するパターンに従います。
ソースドキュメントの例
{
"_id": "62cd344c85c1ea6a2a9f****",
"person": {
"name": "neo",
"age": 26,
"sex": "male"
}
}送信先テーブル
| 列名 | 型 | 備考 |
|---|---|---|
mongo_id | VARCHAR | プライマリキー |
person_name | VARCHAR | — |
person_age | DECIMAL | — |
列の割り当て
| 列名 | 型 | 割り当て式 |
|---|---|---|
mongo_id | STRING | bson_value("_id") |
person_name | STRING | bson_value("person","name") |
person_age | DECIMAL | bson_value("person","age") |
親フィールドのみ(例:bson_value("person"))を指定し、bson_value("person","name") のように完全なフィールドパスを指定しないと、すべてのサブフィールド(name、age、sex)の増分データが失われる可能性があります。必ず完全なフィールドパスを指定してください。
次のステップ
複数テーブルのマージ機能を有効化する — 複数のソーステーブルを単一の送信先テーブルにマージします
オブジェクト名のマッピング — 送信先でコレクションまたはデータベースの名前を変更します
モニタリングとアラートの設定 — タスクの失敗やレイテンシーしきい値に関するアラートを設定します
データベース接続の管理 — タスク間で再利用可能なデータベースインスタンスの登録