Tutorial: Copiar datos continuamente de MongoDB a AWS S3

Despliega una instancia de Base de Datos Federada con S3 y los almacenes de datos Atlas.

Una instancia federada de base de datos consolida múltiples fuentes de datos en una única interfaz consultable. En este tutorial, conectará su bucket S3 y su clúster de Atlas como almacenes de datos en la misma instancia federada de base de datos. La conexión de ambos almacenes de datos permite que el activador de copia lea del clúster y guarde en S3.

1. Implementar una instancia federada de base de datos con un almacén de datos S3. Para aprender cómo, consulte Implementar un almacén de datos de instancia federada de base de datos. Cuando configure el almacén de datos S3, otorgue acceso al bucket al rol de IAM Read and write para que Atlas Data Federation pueda guardar archivos Parquet.

2. Agrega tu clúster de Atlas como un segundo almacén de datos en la instancia federada de base de datos.

Después de completar estos pasos, anote el nombre de su servicio de instancia federada de base de datos. Necesitarás este nombre en un paso posterior.

Crea un activador programado para insertar documentos de prueba.

Crea un activador programado que inserte un nuevo documento en tu clúster cada minuto. Esto genera datos de prueba para que puedas verificar que el activador de copia funciona.

1. En Atlas, diríjase a la página Triggers.

   1. Si aún no aparece, se debe seleccionar la organización que contiene el proyecto en el menú Organizations de la barra de navegación.

   2. Si aún no se muestra, seleccione su proyecto en el menú Projects de la barra de navegación.

   3. En la barra lateral, haz clic en Triggers en la sección Streaming Data.

   Se muestra la página de Triggers.

2. Haga clic en Add Trigger.

3. Selecciona Scheduled como el Trigger Type.

4. En Trigger Details, configure lo siguiente:

   Configuración	Valor
   Trigger Name	Create_Event_Every_Min_Trigger
   Schedule Type	Basic
   Interval	Cada 1 minuto
   Event Type	Function

5. En la sección Function, selecciona + New Function e ingresa el siguiente código. Reemplace los valores de los marcadores de posición con los nombres de su servicio Atlas, base de datos y colección.

   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. Haga clic en Save.

   Después de ejecutar el activador, confirma que aparezcan nuevos documentos en tu colección de clústeres cada minuto.

Crea un desencadenador programado para copiar datos a S3.

Cree un activador programado que ejecute un pipeline de agregación utilizando la etapa $out para copiar documentos recientes de su clúster a su bucket S3 en formato Parquet cada minuto.

1. En la página Triggers, haz clic en Add Trigger.

2. Selecciona Scheduled como el Trigger Type.

3. En Trigger Details, configure lo siguiente:

   Configuración	Valor
   Trigger Name	Copy_Events_To_S3_Trigger
   Schedule Type	Basic
   Interval	Cada 1 minuto
   Event Type	Function

4. En la sección Function, seleccione + New Function e introduzca el siguiente código. Reemplace los valores de marcador de posición con los nombres de su servicio de instancia federada de base de datos, base de datos virtual, colección virtual, depósito S3 y región de 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. Haga clic en Save.

   Después de que se ejecute el activador, confirme que aparece un archivo Parquet llamado events en su bucket S3.