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

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

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

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 ステージを使用して、クラスターから最新のドキュメントを Parquet 形式でクラウドプロバイダーの S3 バケットに 1 分ごとにコピーします。

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 バケット内に表示されることを確認します。