Data Transmission Service:ApsaraDB for MongoDB から AnalyticDB for PostgreSQL へのデータ同期

前提条件

• ターゲットの AnalyticDB for PostgreSQL インスタンスを作成済みであること。そのストレージ容量は、ソースの ApsaraDB for MongoDB インスタンスが使用するストレージよりも大きい必要があります。詳細については、「インスタンスの作成」をご参照ください。

  説明

  ターゲットインスタンスのストレージ容量は、ソースデータベースが使用するストレージより少なくとも 10% 大きくする必要があります。

• ターゲットの AnalyticDB for PostgreSQL インスタンスで、データベース、スキーマ、およびプライマリキーを持つテーブルを作成済みであること。詳細については、「SQL 構文」をご参照ください。

  重要

  • ターゲットテーブルのデータ型が、ソースの MongoDB データと互換性があることを確認してください。たとえば、MongoDB の _id フィールドが ObjectId 型の場合、AnalyticDB for PostgreSQL インスタンスの対応するデータ型は varchar である必要があります。

  • AnalyticDB for PostgreSQL インスタンスのターゲットテーブルには、_id または _value という名前の列を含めることはできません。

• ソースの ApsaraDB for MongoDB インスタンスがシャードクラスターの場合、すべてのシャードノードのエンドポイントを取得する必要があります。各シャードのデータベースアカウントとパスワードは同じである必要があります。詳細については、「シャードエンドポイントの申請」をご参照ください。

注意事項

課金

同期タイプ

データベースアカウントの権限

操作手順

1. ターゲットリージョンの同期タスクリストページに移動します。次の 2 つの方法のいずれかを使用できます：

   DTS コンソールから

   1. Data Transmission Service (DTS) コンソールにログインします。

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

   3. ページの左上隅で、同期インスタンスが配置されているリージョンを選択します。

   DMS コンソールから

   説明

   実際の操作は、DMS コンソールのモードとレイアウトによって異なる場合があります。詳細については、「シンプルモード」および「DMS インターフェイスのレイアウトとスタイルをカスタマイズする」をご参照ください。

   1. Data Management (DMS) にログインします。

   2. トップメニューバーで、[データ + AI] > [Data Transmission (DTS)] > [データ同期] を選択します。

   3. データ同期タスク の右側で、同期インスタンスが配置されているリージョンを選択します。

2. タスクの作成 をクリックして、タスク設定ページを開きます。

3. ソースデータベースとターゲットデータベースを設定します。

   カテゴリ	パラメーター	説明
   N/A	タスク名	DTS は自動的にタスク名を生成します。識別しやすいように、わかりやすい名前を指定することを推奨します。名前は一意である必要はありません。
   移行元データベース	既存の接続情報の選択	システムに追加された (新しく作成または保存された) データベースインスタンスを使用するには、ドロップダウンリストからデータベースインスタンスを選択します。データベース情報は自動的に設定されます。 説明 DMS コンソールでは、この設定項目は DMS データベースインスタンスの選択 という名前です。 データベースインスタンスをシステムに追加していない場合、またはすでに追加されているインスタンスを使用する必要がない場合は、次のデータベース情報を手動で設定します。
   	データベースタイプ	MongoDB を選択します。
   	アクセス方法	Alibaba Cloud インスタンス を選択します。
   	インスタンスのリージョン	ソースの ApsaraDB for MongoDB インスタンスが配置されているリージョンを選択します。
   	Alibaba Cloud アカウント間でデータを複製	この例では、現在の Alibaba Cloud アカウントに属するデータベースインスタンスを使用します。× を選択します。
   	アーキテクチャ	この例では、レプリカセット を選択します。 説明 ソースの ApsaraDB for MongoDB インスタンスが シャードクラスター アーキテクチャを使用している場合、Shardアカウント と Shardパスワード も指定する必要があります。
   	移行方法	要件に基づいて、増分データ同期の方法を選択します。 Oplog (推奨)： このオプションは、ソースデータベースで Oplog が有効になっている場合に使用できます。 説明 Oplog は、自己管理の MongoDB データベースおよび ApsaraDB for MongoDB インスタンスでデフォルトで有効になっています。この方法では、ログの取得が速いため、増分同期タスクのレイテンシが低くなります。Oplog を選択することを推奨します。 ChangeStream： このオプションは、ソースデータベースで変更ストリームが有効になっている場合に使用できます。 説明 ソースデータベースが Amazon DocumentDB (非エラスティッククラスター) インスタンスの場合、ChangeStream のみを選択できます。 ソースデータベースの アーキテクチャ を シャードクラスター に設定した場合、Shardアカウント と Shardパスワード を指定する必要はありません。
   	インスタンス ID	ソースの ApsaraDB for MongoDB インスタンスの ID を選択します。
   	認証データベース	ソースの ApsaraDB for MongoDB インスタンスのデータベースアカウントが属するデータベースの名前を入力します。名前を変更していない場合、デフォルト値は admin です。
   	データベースアカウント	ソースの ApsaraDB for MongoDB インスタンスのデータベースアカウントを入力します。必要な権限については、「データベースアカウントに必要な権限」をご参照ください。
   	データベースのパスワード	データベースアカウントに対応するパスワードを入力します。
   	暗号化	DTS は 非暗号化、SSL 暗号化、Mongo Atlas SSL の 3 つの接続方法をサポートしています。暗号化 のオプションは、選択した アクセス方法 と アーキテクチャ によって異なります。コンソールに表示されるオプションが優先されます。 説明 アーキテクチャ が シャードクラスター で、移行方法 が Oplog の MongoDB データベースは、SSL 暗号化 をサポートしていません。 ソースが自己管理の MongoDB データベース (アクセス方法 が Alibaba Cloud インスタンス ではない) で、レプリカセット アーキテクチャを使用し、SSL 暗号化 を選択した場合、DTS は接続を検証するために CA 証明書をアップロードすることもできます。
   移行先データベース	既存の接続情報の選択	システムに追加された (新しく作成または保存された) データベースインスタンスを使用するには、ドロップダウンリストからデータベースインスタンスを選択します。データベース情報は自動的に設定されます。 説明 DMS コンソールでは、この設定項目は DMS データベースインスタンスの選択 という名前です。 データベースインスタンスをシステムに追加していない場合、またはすでに追加されているインスタンスを使用する必要がない場合は、次のデータベース情報を手動で設定します。
   	データベースタイプ	AnalyticDB for PostgreSQL を選択します。
   	アクセス方法	Alibaba Cloud インスタンス を選択します。
   	インスタンスのリージョン	ターゲットの AnalyticDB for PostgreSQL インスタンスが配置されているリージョンを選択します。
   	インスタンス ID	ターゲットの AnalyticDB for PostgreSQL インスタンスの ID を選択します。
   	データベース名	同期されたオブジェクトを格納するターゲットの AnalyticDB for PostgreSQL インスタンスのデータベースの名前を入力します。
   	データベースアカウント	ターゲットの AnalyticDB for PostgreSQL インスタンスのデータベースアカウントを入力します。必要な権限については、「データベースアカウントに必要な権限」をご参照ください。
   	データベースのパスワード	データベースアカウントに対応するパスワードを入力します。

4. 設定が完了したら、ページ下部の 接続をテストして続行 をクリックします。

   説明

   • DTS サーバーからのアクセスを許可するために、DTS サーバーの IP アドレス CIDR ブロックがソースおよびターゲットデータベースのセキュリティ設定に追加されていることを確認してください。これは自動または手動で行うことができます。詳細については、「DTS サーバーの IP アドレス CIDR ブロックをホワイトリストに追加する」をご参照ください。

   • ソースまたはターゲットデータベースが自己管理データベース (アクセス方法 が Alibaba Cloud インスタンス ではない) の場合、DTS サーバーの CIDR ブロック ダイアログボックスで 接続テスト をクリックする必要もあります。

5. タスクオブジェクトを設定します。

   1. オブジェクト設定 ページで、同期するオブジェクトを設定します。

      パラメーター	説明
      同期タイプ	増分データ同期 が選択されています。完全データ同期 のみを選択できます。スキーマ同期 はサポートされていません。事前チェックが完了すると、DTS は選択されたオブジェクトのデータをソースインスタンスからターゲットクラスターに同期します。これは、後続の増分データ同期のベースラインデータとして機能します。
      同期する DDL および DML 操作	インスタンスレベルで同期する増分操作を選択します。 説明 コレクションレベルで増分操作を選択するには、選択中のオブジェクト セクションで同期オブジェクトを右クリックし、表示されるダイアログボックスで目的の操作を選択します。
      競合するテーブルの処理モード	エラーの事前チェックと報告：ターゲットデータベースに同じ名前のテーブルが存在するかどうかを確認します。同じ名前のテーブルが存在しない場合、事前チェックは成功します。同じ名前のテーブルが存在する場合、事前チェックは失敗し、データ同期タスクは開始されません。 説明 ターゲットデータベースで同じ名前のテーブルを削除または名前変更できない場合は、別のテーブル名にマッピングできます。詳細については、「テーブル名と列名のマッピング」をご参照ください。 エラーを無視して続行：ターゲットデータベースでの重複テーブル名のチェックをスキップします。 警告 エラーを無視して続行 を選択すると、データの不整合が発生し、ビジネスにリスクをもたらす可能性があります。例： テーブルスキーマが同じで、ターゲットデータベースのレコードがソースデータベースのレコードと同じプライマリキーまたは一意キーの値を持つ場合： 完全同期中、DTS はターゲットクラスターのレコードを保持します。ソースデータベースの対応するレコードは同期されません。 増分同期中、ソースデータベースのレコードがターゲットデータベースのレコードを上書きします。 テーブルスキーマが異なる場合、初期データ同期が失敗する可能性があります。これにより、一部の列データのみが同期されるか、同期が完全に失敗する可能性があります。注意して進めてください。
      ソースオブジェクト	ソースオブジェクト ボックスで、同期するオブジェクトをクリックし、 をクリックして 選択中のオブジェクト ボックスに移動します。 説明 同期するオブジェクトはコレクションレベルで選択できます。
      選択中のオブジェクト	データベース名のマッピングを編集します。 選択中のオブジェクト ボックスで、同期するコレクションを含むソースデータベースを右クリックします。 データベース名 を、ターゲットの AnalyticDB for PostgreSQL インスタンスでデータを受信するスキーマの名前に変更します。表示される [データベースの編集] ダイアログボックスで、[データベース名] (ターゲットスキーマの名前) を変更します。 (オプション) 同期する DDL および DML 操作を選択 セクションで、同期する DML 操作 (insert、update、delete) を選択します。 確かですか をクリックします。 テーブル名のマッピングを編集します。 選択中のオブジェクト ボックスで、同期するコレクションを右クリックします。 [選択されたオブジェクト] パネルには、オブジェクトがツリー構造で表示されます。レベルは、ターゲットデータベース名 (例：dtstest) > テーブル > 特定のテーブル名 (例：class) です。後続の操作を行うには、ターゲットテーブルを右クリックします。 テーブル名 を、ターゲットの AnalyticDB for PostgreSQL インスタンスでデータを受信するテーブルの名前に変更します。表示される [テーブルの編集] ダイアログボックスで、[テーブル名] (例：class) を変更します。テーブル名または列名を編集すると、ターゲットデータベースのテーブル名と列名が新しい名前に変更されます。 (オプション) フィルター条件を指定します。詳細については、「フィルター条件の設定」をご参照ください。[フィルター条件] テキストボックスに、MongoDB のフィルター条件を入力できます。構文は標準の SQL WHERE 句とは異なります。例：{"_id": {$gt:"user100844658590795****",$lte:"user101674868045948****"}}。ここで、$gt は「より大きい」、$lte は「以下」を意味します。 (オプション) 同期する DDL および DML 操作を選択 セクションで、insert、update、delete を選択します。 MongoDB データベースから同期するフィールドを設定します。 デフォルトでは、DTS は同期するコレクションのデータをマッピングし、代入 列に式を設定します。式が要件を満たしているかどうかを確認し、列名、タイプ、長さ、精度 などのパラメーターを設定する必要があります。 代入 列の bson_value() 式で、MongoDB データベースの対応するフィールド名を表示します。 引用符 ("") 内のフィールド名は、MongoDB のフィールド名です。たとえば、式が bson_value("age") の場合、この行は MongoDB の age フィールドに対応します。 (オプション) 同期する必要のないフィールドを削除します。 説明 同期する必要のないフィールドを削除するには、そのフィールドの行の末尾にある アイコンをクリックします。 同期するフィールドを設定します。 bson_value() 式が要件を満たしているかどうかに基づいて、次のいずれかの手順を実行します。 正しい式 列名 を入力します。 説明 これは、ターゲットの AnalyticDB for PostgreSQL インスタンスでデータを受信する列の名前です。 列のデータ タイプ を選択してください。 重要 ターゲットテーブルのデータ型が、ソースの MongoDB データベースのデータと互換性があることを確認してください。 (オプション) 列データの 長さ と 精度 を指定します。 前の手順を繰り返して、必要な各フィールドをマッピングします。 正しくない式 説明 たとえば、親子構造などの階層関係を持つフィールド。 操作 列で、そのフィールドの行の末尾にある アイコンをクリックします。 [+ 列を追加] をクリックします。[追加の列] セクションで、右上隅の [+ 列を追加] をクリックします。 列名、タイプ、長さ、精度 パラメーターを設定します。 代入 の下のテキストボックスに、bson_value() 式を入力します。詳細については、「値の設定例」をご参照ください。 重要 ターゲットテーブルのプライマリキー列に bson_value("_id") を割り当てる必要があります。 bson_value() 式を設定する際は、階層に基づいて最小のサブフィールドまでのパスを指定する必要があります。そうしないと、データ損失やタスクの失敗が発生する可能性があります。 前の手順を繰り返して、必要な各フィールドをマッピングします。 確かですか をクリックします。

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

      パラメーター	説明
      タスクのスケジュールに使用する専用クラスターの選択	デフォルトでは、DTS は共有クラスターでタスクをスケジュールするため、クラスターを選択する必要はありません。より安定したパフォーマンスを得るために、DTS 同期タスクを実行するための専用クラスターを購入できます。詳細については、「DTS 専用クラスターとは」をご参照ください。
      失敗した接続の再試行時間	同期タスクの開始後、ソースまたはターゲットデータベースへの接続が失敗した場合、DTS はエラーを報告し、直ちに接続の再試行を開始します。デフォルトの再試行時間は 720 分です。10 分から 1,440 分の範囲でカスタムの再試行時間を指定することもできます。30 分以上に設定することを推奨します。指定された時間内に DTS がデータベースに正常に再接続した場合、同期タスクは自動的に再開されます。そうでない場合、タスクは失敗します。 説明 同じソースまたはターゲットを共有する複数の DTS インスタンス (例：インスタンス A とインスタンス B) があり、インスタンス A のネットワーク再試行時間を 30 分、インスタンス B を 60 分に設定した場合、両方に短い方の 30 分が使用されます。 DTS は接続再試行期間中のタスク実行時間に対して課金するため、ビジネスニーズに基づいて再試行時間をカスタマイズするか、ソースおよびターゲットデータベースインスタンスがリリースされた後、できるだけ早く DTS インスタンスをリリースすることを推奨します。
      移行元データベースと移行先データベースで他の問題が発生した場合の、再試行までの待機時間です。	同期タスクの開始後、ソースまたはターゲットデータベースで接続以外の問題 (DDL または DML 実行例外など) が発生した場合、DTS はエラーを報告し、直ちに連続的な再試行操作を開始します。デフォルトの再試行時間は 10 分です。1 分から 1,440 分の範囲でカスタムの再試行時間を指定することもできます。10 分以上に設定することを推奨します。設定された再試行時間内に関連操作が成功した場合、同期タスクは自動的に再開されます。そうでない場合、タスクは失敗します。 重要 移行元データベースと移行先データベースで他の問題が発生した場合の、再試行までの待機時間です。 の値は、失敗した接続の再試行時間 の値より小さくする必要があります。
      完全同期レートを制限するかどうか	完全同期段階では、DTS はソースおよびターゲットデータベースの読み取りおよび書き込みリソースを消費し、データベースの負荷を増加させる可能性があります。ソースおよびターゲットデータベースの負荷を軽減するために、1 秒あたりのソースデータベースのクエリ率 QPS、1 秒あたりの完全移行の行数 RPS、および 1 秒あたりの完全移行データ量 (MB) BPS パラメーターを設定することで、完全同期タスクのレート制限を設定できます。 説明 この設定項目は、同期タイプ が 完全データ同期 に設定されている場合にのみ使用できます。 同期インスタンスの実行後に、完全同期レートを調整することもできます。
      同期するデータのうち、同一テーブル内のプライマリキー_id のデータ型が一意かどうか	同期する各コレクション内で、_id プライマリキーのデータ型が一意であるかどうかを指定します。 重要 実際のデータに基づいてオプションを選択してください。誤った選択はデータ損失につながる可能性があります。 このパラメーターは、同期タイプ に 完全データ同期 が含まれている場合にのみ使用できます。 ○：データ型は一意です。完全同期段階では、DTS はソースデータのプライマリキーのデータ型をスキャンしません。各コレクションについて、DTS は 1 つのプライマリキーデータ型に対応するデータのみを同期します。 ×：データ型は一意ではありません。完全同期段階では、DTS はソースデータのプライマリキーのデータ型をスキャンし、すべてのデータを同期します。
      増分同期率を制限するかどうか	増分同期タスクにもレート制限を設定できます。ターゲットデータベースへの圧力を軽減するために、1 秒あたりの増分同期の行数 RPS と 1 秒あたりの増分同期データ量 (MB) BPS を設定します。
      環境タグ	ビジネスニーズに基づいてインスタンスを識別するために環境タグを選択できます。この例では、タグは必要ありません。
      ETL 機能の設定	抽出、変換、ロード (ETL) 機能を有効にするかどうかを選択します。詳細については、「ETL とは」をご参照ください。有効な値： ○：ETL 機能を有効にします。コードエディタにデータ処理文を入力します。詳細については、「データ移行またはデータ同期タスクで ETL を設定する」をご参照ください。 ×：ETL 機能を無効にします。
      監視アラート	アラートを設定するかどうかを指定します。同期が失敗した場合や、レイテンシが指定されたしきい値を超えた場合に、アラート連絡先に通知が送信されます。 ×：アラートは設定されません。 ○：アラートのしきい値を設定し、アラート通知を指定してアラートを設定します。詳細については、「タスク設定中にモニタリングとアラートを設定する」をご参照ください。

6. タスクを保存し、事前チェックを実行します。

   • このインスタンスを設定するための API パラメーターを表示するには、次：タスク設定の保存と事前チェック ボタンにカーソルを合わせ、バブル内の OpenAPI パラメーターのプレビュー をクリックします。

   • API パラメーターの表示が完了したら、ページ下部の 次：タスク設定の保存と事前チェック をクリックします。

   説明

   • 同期ジョブが開始される前に、DTS は事前チェックを実行します。ジョブは、すべての事前チェック項目が成功した後にのみ開始できます。

   • 事前チェックが失敗した場合は、失敗した項目の 詳細を表示 をクリックします。プロンプトに従って問題を修正し、再度事前チェックを実行します。

   • 事前チェックで警告が返された場合：

     • 確認項目が失敗し、無視できない場合は、項目の横にある 詳細を表示 をクリックします。指示に従って問題を修正し、再度事前チェックを実行します。

     • 無視できる確認項目については、アラートの詳細を確認、無視、OK、再度事前チェックを実行 を順にクリックして、警告をスキップし、再度事前チェックを実行できます。警告項目を無視することを選択した場合、データの不整合などの問題が発生し、ビジネスにリスクをもたらす可能性があります。

7. インスタンスを購入します。

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

   2. 購入 ページで、データ同期インスタンスの課金方法とリンク仕様を選択します。次の表で、これらのパラメーターについて詳しく説明します。

      カテゴリ	パラメーター	説明
      新しいインスタンスクラス	課金方法	サブスクリプション：インスタンス作成時に支払います。これは長期的なニーズに適しており、従量課金よりもコスト効率が高いです。サブスクリプション期間が長いほど、割引率が高くなります。 従量課金：時間単位で課金されます。これは短期的なニーズに適しています。使用後すぐにインスタンスをリリースしてコストを節約できます。
      	リソースグループの設定	インスタンスが属するリソースグループ。デフォルトはデフォルトリソースグループです。詳細については、「Resource Management とは」をご参照ください。
      	リンク仕様	DTS は、異なるパフォーマンスレベルの同期仕様を提供します。同期リンク仕様は同期レートに影響します。ビジネスシナリオに基づいて仕様を選択できます。詳細については、「データ同期リンクの仕様」をご参照ください。
      	サブスクリプション期間	サブスクリプションモードでは、サブスクリプションインスタンスの期間と数量を選択します。1 か月から 9 か月までの月次サブスクリプション、または 1、2、3、5 年の年次サブスクリプションを選択できます。 説明 このオプションは、課金方法が サブスクリプション の場合にのみ使用できます。

   3. 設定が完了したら、Data Transmission Service (従量課金) 利用規約 を読み、選択します。

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

      [データ同期] ページでタスクの進捗状況を確認できます。

値の割り当て例

{
  "_id":"62cd344c85c1ea6a2a9f****",
  "person":{
    "name":"neo",
    "age":26,
    "sex":"male"
  }
}