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

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

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

1. S3 データストアを使用して、フェデレーティッドデータベースインスタンスを配置します。方法については、「フェデレーティッドデータベースインスタンスデータ保存 (する)の配置」を参照してください。S3 データストアを設定する際には、IAM ロール Read and write にバケットへのアクセス権を付与します。これにより、Atlas Data Federation は Parquet ファイルを書き込めるようになります。

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 ステージを使用して集計パイプラインを実行し、クラスターから最新のドキュメントをおその S3 バケットに Parquet 形式でコピーする予定されたトリガーを作成します。

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 という名前の Parquet ファイルが、ユーザーの S3 バケットに表示されることを確認します。