Tutorial: Copiar continuamente dados do MongoDB para o AWS S3

Implemente uma Instância do Banco de Dados Federado com S3 e armazenamentos de dados do Atlas .

Uma instância do banco de dados federado consolida várias fontes de dados em uma única interface para query. Neste tutorial, você conecta seu bucket S e seu cluster Atlas como armazenamentos de dados na mesma instância do banco de dados federado. A3 conexão de ambos os armazenamentos de dados permite que o trigger de cópia leia do cluster e grave em3 S.

1. Implemente uma instância do banco de dados federado com um armazenamento de 3 dados S. Para saber como, consulte Implantar um Armazenamento de Dados de Instância de Banco de Dados Federado. Ao configurar o armazenamento de dados S, conceda à função do 3 IAM Read and write acesso ao bucket para que o Atlas Data Federation possa gravar arquivos Parquet.

2. Adicione seu Atlas cluster como um segundo armazenamento de dados na instância do banco de dados federado.

Depois de concluir estas etapas, anote o nome do serviço de instância do banco de dados federado . Você precisará desse nome em uma etapa posterior.

Crie um gatilho programado para inserir documentos de teste.

Crie um Trigger Agendado que insira um novo documento em seu cluster a cada minuto. Isso gera dados de teste para que você possa verificar se o trigger de cópia funciona.

1. No Atlas, acesse a página Triggers.

   1. Se ainda não tiver sido exibido, selecione a organização que contém seu projeto no menu Organizations na barra de navegação.

   2. Se ainda não estiver exibido, selecione seu projeto no menu Projects na barra de navegação.

   3. Na barra lateral, clique em Triggers sob o título Streaming Data.

   A página Acionadores é exibida.

2. Clique em Add Trigger.

3. Selecione Scheduled como Trigger Type.

4. No Trigger Details, defina a seguinte configuração:

   Contexto	Valor
   Trigger Name	Create_Event_Every_Min_Trigger
   Schedule Type	Basic
   Interval	A cada 1 minuto
   Event Type	Function

5. Na seção Function, selecione + New Function e insira o código a seguir. Substitua os valores do espaço reservado pelos nomes do seu serviço, banco de dados e coleção do 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. Clique em Save.

   Após a execução do trigger, confirme que novos documentos aparecem em sua coleção de cluster a cada minuto.

Crie um scheduled trigger para copiar dados para S3.

Crie um Trigger Agendado que execute um pipeline de agregação usando o estágio $out para copiar documentos recentes do cluster para o bucket S3 no formato Parquet a cada minuto.

1. Na página Triggers, clique em Add Trigger.

2. Selecione Scheduled como Trigger Type.

3. No Trigger Details, defina a seguinte configuração:

   Contexto	Valor
   Trigger Name	Copy_Events_To_S3_Trigger
   Schedule Type	Basic
   Interval	A cada 1 minuto
   Event Type	Function

4. Na Function seção, selecione + New Function e insira o código a seguir. Substitua os valores de espaço reservado pelos nomes do serviço de instância do banco de dados federado , banco de dados virtual , coleção virtual, bucketS e3 região do 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. Clique em Save.

   Após a execução do trigger, confirme se um arquivo Parquet chamado events aparece em seu bucket 3 S.