チュートリアル: MongoDBからAmazon Web Services S3 へのデータの継続的なコピー

S3 と Atlas データストアを使用してフェデレーティッドデータベースインスタンスを配置する。

フェデレーティッドデータベースインスタンスは、複数のデータソースを単一のクエリ可能なインターフェースに統合します。このチュートリアルでは、 S3バケットと Atlas クラスターを同じフェデレーティッドデータベースインスタンス内のデータ ストアとして接続します。両方のデータ ストアを接続すると、コピー trigger はクラスターから読み取り、S3 に書込むことができます。

1. Deploy a federated database instance with an S3 data store. To learn how, see Deploy a Federated Database Instance Data Store. When you configure the S3 data store, grant the IAM role Read and write access to the bucket so that Atlas Data Federation can write Parquet files.

2. Atlas クラスターをフェデレーティッドデータベースインスタンスの 2 番目のデータストアとして追加します。

これらの手順が完了したら、フェデレーティッドデータベースインスタンスサービスの名前を控えておきます。この名前は後の手順で必要になります。

テストドキュメントを挿入するための予定されたトリガーを作成します。

毎分ごとに新しいドキュメントをクラスターに挿入する予定されたトリガーを作成します。これによりテスト データが生成され、コピー trigger が動作することを確認できます。

1. Atlas で、 Triggers ページに移動します。

   1. まだ表示されていない場合は、プロジェクトを含む組織をナビゲーション バーの Organizations メニューで選択します。

   2. まだ表示されていない場合は、ナビゲーション バーの Projects メニューからプロジェクトを選択します。

   3. サイドバーで、 Streaming Data見出しの下のTriggersをクリックします。

   Triggersページが表示されます。

2. [Add Trigger] をクリックします。

3. Scheduled を Trigger Type として選択します。

4. Trigger Details で、次の構成を設定します。

   設定	値
   Trigger Name	Create_Event_Every_Min_Trigger
   Schedule Type	Basic
   Interval	1 分ごと
   Event Type	Function

5. Functionセクションで+ New Functionを選択し、次のコードを入力します。プレースホルダー値を、Atlas サービス、データベース、コレクションの名前に置き換えます。

   exports = function () {
      const mongodb = context.services.get(
         "NAME_OF_YOUR_ATLAS_SERVICE"
      );
      const db = mongodb.db("NAME_OF_YOUR_DATABASE");
      const events = db.collection(
         "NAME_OF_YOUR_COLLECTION"
      );
      const event = events.insertOne({
         time: new Date(),
         aNumber: Math.random() * 100,
         type: "event"
      });
      return JSON.stringify(event);
   };

6. [Save] をクリックします。

   trigger の実行後、クラスター コレクションに新しいドキュメントが毎分表示されることを確認します。

S3にデータをコピーする予定されたトリガーを作成します。

$out ステージを使用して集計パイプラインを実行する予定されたトリガーを作成し、1 分ごとに最近のドキュメントをクラスターから S3バケットにコピーし形式。

1. Triggersページで、 Add Triggerをクリックします。

2. Scheduled を Trigger Type として選択します。

3. Trigger Details で、次の構成を設定します。

   設定	値
   Trigger Name	Copy_Events_To_S3_Trigger
   Schedule Type	Basic
   Interval	1 分ごと
   Event Type	Function

4. Function セクションで、+ New Function を選択し、次のコードを入力します。プレースホルダー値を、フェデレーティッドデータベースインスタンスサービス、 仮想データベース、 仮想コレクション、 S3バケット、およびAWSリージョンの名前に置き換えます。

   exports = function () {
      const service = context.services.get(
         "NAME_OF_YOUR_FEDERATED_DATA_SERVICE"
      );
      const db = service.db(
         "NAME_OF_YOUR_VIRTUAL_DATABASE"
      );
      const events = db.collection(
         "NAME_OF_YOUR_VIRTUAL_COLLECTION"
      );
      const pipeline = [
         {
            $match: {
               "time": {
                  $gt: new Date(
                     Date.now() - 60 * 1000
                  ),
                  $lt: new Date(Date.now())
               }
            }
         },
         {
            "$out": {
               "s3": {
                  "bucket": "YOUR_S3_BUCKET_NAME",
                  "region": "YOUR_AWS_REGION",
                  "filename": "events",
                  "format": {
                     "name": "parquet",
                     "maxFileSize": "10GB",
                     "maxRowGroupSize": "100MB"
                  }
               }
            }
         }
      ];
      return events.aggregate(pipeline);
   };

5. [Save] をクリックします。

   trigger が実行された後、events という名前の PerquetファイルがS3バケットに表示されることを確認します。