Tutorial: Copiar continuamente dados do MongoDB para o Amazon Web Services S3

Implante 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 consultável. Neste tutorial, você conecta seu bucket S3 e seu cluster Atlas como armazenamentos de dados na mesma instância do banco de dados federado. A conexão de ambos os armazenamentos de dados permite que o trigger de cópia leia do cluster e grave em S3.

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

2. Adicione seu cluster Atlas 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 da instância do banco de dados federado. Você precisará desse nome em uma etapa posterior.

Crie um trigger agendado para inserir documentos de teste.

Crie um trigger agendado que insere um novo documento no 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 seguinte código. Substitua os valores do espaço reservado pelos nomes do 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.

   Depois que o trigger for executado, confirme se novos documentos aparecem na coleção do cluster a cada minuto.

Crie um trigger agendado 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 seu cluster para o seu 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 seção Function, 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, bucket S3 e 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.

   Depois que o trigger for executado, confirme se um arquivo Parquet chamado events aparece no seu bucket S3.