Data Transmission Service:ApsaraDB for MongoDB のシャーディングクラスターインスタンスから Function Compute の関数へのデータ同期

関数が受信する内容

各呼び出しは、Records 配列を持つオブジェクトを渡します。配列内のすべての要素は、1 つの変更イベントを表します。

関数は 2 つのカテゴリの操作を受信します。

• DDL — スキーマ変更： CreateIndex、CreateCollection、DropIndex、DropCollection

• DML — データ変更： INSERT、UPDATE、DELETE

db.createCollection("testCollection")

{
  "Records": [{
    "data": [{"doc": "{\"create\": \"testCollection\", \"idIndex\": {\"v\": 2, \"key\": {\"_id\": 1}, \"name\": \"_id_\"}}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056437000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "testCollection",
    "ts": 1694056437510
  }]
}

db.testCollection.drop()

{
  "Records": [{
    "data": [{"doc": "{\"drop\": \"testCollection\"}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056577000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "testCollection",
    "ts": 1694056577789
  }]
}

db.testCollection.createIndex({name: 1})

{
  "Records": [{
    "data": [{"doc": "{\"createIndexes\": \"testCollection\", \"v\": 2, \"key\": {\"name\": 1}, \"name\": \"name_1\"}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056670000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "testCollection",
    "ts": 1694056670719
  }]
}

db.testCollection.dropIndex({name: 1})

{
  "Records": [{
    "data": [{"doc": "{\"dropIndexes\": \"testCollection\", \"index\": \"name_1\"}"}],
    "pkNames": ["_id"],
    "type": "DDL",
    "es": 1694056817000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": true,
    "table": "$cmd",
    "ts": 1694056818035
  }]
}

// バッチ挿入
db.runCommand({insert: "user", documents: [{"name": "jack", "age": 20}, {"name": "lili", "age": 20}]})

// 単一挿入
db.user.insert({"name": "jack", "age": 20})

{
  "Records": [
    {
      "data": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}, \"name\": \"jack\", \"age\": 20}"}],
      "pkNames": ["_id"],
      "type": "INSERT",
      "es": 1694054783000,
      "database": "MongoDBTest",
      "id": 0,
      "isDdl": false,
      "table": "user",
      "ts": 1694054784427
    },
    {
      "data": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}, \"name\": \"lili\", \"age\": 20}"}],
      "pkNames": ["_id"],
      "type": "INSERT",
      "es": 1694054783000,
      "database": "MongoDBTest",
      "id": 0,
      "isDdl": false,
      "table": "user",
      "ts": 1694054784428
    }
  ]
}

db.user.update({"name": "jack"}, {$set: {"age": 30}})

{
  "Records": [{
    "data": [{"doc": "{\"$set\": {\"age\": 30}}"}],
    "pkNames": ["_id"],
    "old": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}}"}],
    "type": "UPDATE",
    "es": 1694054989000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": false,
    "table": "user",
    "ts": 1694054990555
  }]
}

db.user.remove({"name": "jack"})

{
  "Records": [{
    "data": [{"doc": "{\"_id\": {\"$oid\": \"64f9397f6e255f74d65a****\"}}"}],
    "pkNames": ["_id"],
    "type": "DELETE",
    "es": 1694055452000,
    "database": "MongoDBTest",
    "id": 0,
    "isDdl": false,
    "table": "user",
    "ts": 1694055452852
  }]
}

制限事項

同期タスクを作成する前に、これらの制約を確認してください。

スコープの制約：

• 増分データ同期のみがサポートされます。フルデータ同期はサポートされません。

• クロスリージョン同期はサポートされません。

• DTS は、admin、config、または local データベースからデータを同期できません。

• オブジェクトマッピングはサポートされません。

• トランザクション情報は保持されません。トランザクションは、宛先で個別のレコードに変換されます。

ソースデータベースの制約：

• ソースはシャーディングクラスターアーキテクチャを使用する必要があり、Mongos ノードは 10 個以下である必要があります。

• ソースを Azure Cosmos DB for MongoDB クラスターまたは Amazon DocumentDB エラスティッククラスターにすることはできません。

• 同期するコレクションには、プライマリキーまたは UNIQUE 制約が必要で、すべてのフィールドが一意である必要があります。そうでない場合、宛先データベースに重複したデータレコードが含まれる可能性があります。

• 単一のドキュメントは 16 MB を超えることはできません。このサイズを超えるドキュメントは宛先の関数に書き込むことができず、エラーを引き起こします。必要に応じて、抽出、変換、ロード (ETL) 機能を使用して大きなフィールドをフィルターします。

• タスクごとに最大 1,000 個のコレクションを同期できます。より多くのコレクションを同期するには、複数のタスクを作成するか、データベースレベルで同期します。

• 同期タスクの実行中は、ソースインスタンスをスケーリングできません。

• DTS は SRV エンドポイント経由で MongoDB データベースに接続できません。

• ソースデータベースのバランサーがアクティブな場合、タスクに遅延が発生することがあります。

• ソースデータベースがシャーディングクラスターアーキテクチャを使用するセルフマネージド MongoDB データベースの場合、[アクセス方法] パラメーターを Express Connect、VPN Gateway、または Smart Access Gateway または Cloud Enterprise Network (CEN) に設定します。

書き込みの制約：

• INSERT 操作の場合、挿入されるデータにはシャードキーが含まれている必要があります。

• UPDATE 操作の場合、シャードキーは変更できません。

• 宛先の関数ごとに 1 つの DTS タスクのみを設定してください。同じ関数に書き込む複数のタスクは、データエラーを引き起こす可能性があります。

oplogと変更ストリームの要件：

• oplog を有効にして少なくとも 7 日間のログデータを保持するか、変更ストリームを有効にして少なくとも過去 7 日間の変更をカバーする必要があります。どちらの条件も満たされない場合、DTS はソースの変更をキャプチャできず、DTS のサービスレベルアグリーメント (SLA) の対象外となるデータの不整合や損失を引き起こす可能性があります。

変更ストリームの制限 (該当する場合)：

• 変更ストリームには MongoDB 4.0 以降が必要です。

• 変更ストリームを使用する場合、双方向同期はサポートされません。

• 非エラスティックな Amazon DocumentDB クラスターの場合は、変更ストリームを使用します。[移行方法] を [ChangeStream] に、[アーキテクチャ] を [シャーディングクラスター] に設定します。

新しいデータベース：

• DTS は、同期タスクの開始後に作成されたデータベースからの増分データを同期しません。

サポートされる操作

キャプチャされる操作は、移行方法によって異なります。

oplog の使用 (推奨)：

• CREATE COLLECTION、CREATE INDEX

• DROP DATABASE、DROP COLLECTION、DROP INDEX

• RENAME COLLECTION

• ドキュメントレベルのINSERT、UPDATE、DELETE

変更ストリームの使用：

• DROP DATABASE、DROP COLLECTION

• RENAME COLLECTION

• ドキュメントレベルのINSERT、UPDATE、DELETE

課金

増分データ同期は課金対象です。料金の詳細については、「課金の概要」をご参照ください。

DTS は、接続リトライ期間中もインスタンスに課金します。

前提条件

開始する前に、以下が準備できていることを確認してください。

• 実行中の ApsaraDB for MongoDB シャーディングクラスターインスタンス。「シャーディングクラスターインスタンスの作成」をご参照ください

• [ハンドラータイプ] が [イベントハンドラー] に設定された Function Compute のサービスと関数。「関数のクイック作成」をご参照ください

• ソースの ApsaraDB for MongoDB インスタンス上のデータベースアカウントに、ソース、admin、および local データベースに対する読み取り権限があること。「MongoDBデータベースユーザーの権限管理」をご参照ください

  説明

  増分同期方法として ChangeStream を使用する場合、ソースデータベースアカウントにはインスタンス全体の Change Streams 読み取り権限 (readAnyDatabase など) が必要です。 ソースがカスタムアカウントを使用する ApsaraDB for MongoDB インスタンスの場合、アカウントに admin データベースに対する読み取り権限も付与する必要があります。 詳細については、「インスタンス作成時に指定する root アカウントの権限」をご参照ください。

同期タスクの作成

ステップ4：同期オブジェクトの選択

[オブジェクトの設定] ステップで、以下を設定します。

パラメーター	説明
[同期タイプ]	[増分データ同期] に固定されています。変更できません
[データ形式]	[Canal Json] に固定されています。フィールドの説明については、「Kafkaクラスターのトピックのデータ形式」の Canal Json セクションをご参照ください
[ソースオブジェクト]	同期するデータベースまたはコレクションを選択し、 をクリックして [選択済みオブジェクト] に移動します
[選択済みオブジェクト]	選択したオブジェクトを確認します。 をクリックしてオブジェクトを削除します。オブジェクトを右クリックして、同期の粒度 (データベースまたはコレクションレベル) を設定します

次のステップ

• Function Compute イベントハンドラーを記述して Canal JSON ペイロードを処理するには、isDdl フィールドを使用して DDL と DML のロジックを分岐させ、各 data 要素の doc 文字列をデシリアライズしてドキュメントフィールドにアクセスします。

• 関数に対する大きなフィールドを持つドキュメントの負荷を軽減するには、データが関数に到達する前に ETL フィルターを設定します。「データ移行またはデータ同期タスクでのETLの設定」をご参照ください。

• 同期の健全性をモニタリングするには、[高度な設定] で、またはタスクの作成後にアラートを設定します。「DTSタスク作成時のモニタリングとアラートの設定」をご参照ください。

• タスクが失敗した場合、DTS のテクニカルサポートが 8 時間以内に復元を試みます。復元中、タスクが再起動されたり、タスクパラメーター (データベースパラメーターではない) が調整されたりすることがあります。変更される可能性のあるパラメーターのリストについては、「インスタンスパラメーターの変更」をご参照ください。