Comenzar con Atlas Stream Processing

Para crear un procesador de flujo en la interfaz de usuario de Atlas, vaya a la página Stream Processing de su proyecto Atlas y haga clic en Manage en el panel de su espacio de trabajo de procesamiento de flujo. A continuación, elija entre usar el constructor visual o el editor JSON para configurar un procesador de flujo llamado solarDemo:

Ejecuta los siguientes comandos en mongosh para crear un procesador de flujo persistente llamado solarDemo:

1. Conéctate a tu espacio de trabajo de stream processing.

   Utiliza la cadena de conexión asociada a tu espacio de trabajo de Stream Processing para conectarte usando mongosh.

   1. En el panel de tu espacio de trabajo de procesamiento de flujos, haz clic en Connect.

   2. En el cuadro de diálogo Connect to your workspace, seleccione la pestaña Shell.

   3. Copie la cadena de conexión que aparece en el cuadro de diálogo. Tiene el siguiente formato, donde <atlas-stream-processing-url> es la URL de tu espacio de trabajo de Stream Processing y <username> es el nombre de usuario de un usuario de base de datos con el rol atlasAdmin:

      mongosh "mongodb://<atlas-stream-processing-url>/"
      --tls --authenticationDatabase admin --username <username>
      --password <password>

   4. Pegue la cadena de conexión en su terminal y reemplace el marcador <password> con las credenciales del usuario.

      Presiona Enter para ejecutarlo y conecta a tu espacio de trabajo de Stream Processing.

2. Configura una etapa $source.

   Defina una variable para una etapa $source que ingiera datos de la fuente sample_stream_solar.

   let s = {
     $source: {
       connectionName: "sample_stream_solar"
     }
   }

3. Configure una etapa $group.

   Defina una variable para una etapa $group que derive la temperatura máxima y los valores promedio, máximo y mínimo de potencia de cada dispositivo solar según su group_id.

   let g = {
     $group: {
       _id: "$group_id",
       max_temp: {
         $max: "$obs.temp"
       },
       avg_watts: {
          $avg: "$obs.watts"
       },
       max_watts: {
         $max: "$obs.watts"
       },
       min_watts: {
         $min: "$obs.watts"
       }
     }
   }

4. Configura una etapa $tumblingWindow.

   Para realizar acumulaciones como $group en datos de streaming, Atlas Stream Processing utiliza ventanas para delimitar el conjunto de datos. Defina una variable para una $tumblingWindow etapa que divida el flujo en 10intervalos consecutivos de segundos.

   Esto significa, por ejemplo, que si la etapa $group calcula un valor para max_watts, extrae el valor máximo de los valores obs.watts de todos los documentos con un group_id dado ingeridos en los 10 segundos anteriores.

    let t = {
      $tumblingWindow: {
        interval: {
          size: NumberInt(10),
          unit: "second"
        },
        pipeline: [g]
     }
   }

5. Configure un escenario de $merge.

   Define una variable para una etapa de $merge que guarda los datos procesados por transmisión a una colección denominada solarColl en la base de datos solarDb de tu clúster conectado de Atlas.

   let m = {
     $merge: {
       into: {
         connectionName: "mongodb1",
         db: "solarDb",
         coll: "solarColl"
       }
     }
   }

6. Crear el procesador de flujo.

   Utilice el método sp.createStreamProcessor() para asignar un nombre a su nuevo procesador de flujo y declarar su pipeline de agregación. La etapa $group pertenece a la pipeline anidada del $tumblingWindow, y no debe incluirse en la definición de pipeline del procesador.

   sp.createStreamProcessor("solarDemo", [s, t, m])

   Esto crea un procesador de flujo llamado solarDemo que aplica la previamente definida query y guarda los datos procesados en la colección solarColl de la base de datos solarDb en el clúster al que se conectó. Devuelve diversas mediciones derivadas de intervalos de observación de 10segundos de tus dispositivos solares.

   Para obtener más información sobre cómo Atlas Stream Processing escribe en bases de datos en reposo, consulta $merge (Stream Processing).

En la lista de procesadores de flujo para tu espacio de trabajo de stream processing, haz clic en el icono Start para tu procesador de flujo.

Utilice el sp.processor.start() método mongosh en:

sp.solarDemo.start()

Para verificar que el procesador de flujo esté desplegando datos en su clúster de Atlas:

1. En Atlas, se debe ir a la página Clusters del proyecto.

   1. Si aún no se muestra, seleccione la organización que contiene su proyecto deseado en el menú Organizations de la barra de navegación.

   2. Si aún no aparece, selecciona el proyecto deseado en el menú Projects de la barra de navegación.

   3. En la barra lateral, haz clic en Clusters en la sección Database.

      La página de clústeres se muestra.

2. En Atlas, se debe ir a la página Data Explorer del proyecto.

   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 Data Explorer en la sección Database.

      El Data Explorer se muestra.

   Nota

   También puede hacer clic en el nombre de un grupo para abrir la barra lateral Cluster y luego hacer clic en Data Explorer debajo del encabezado Shortcuts.

3. Ver la colección MySolar.

Para verificar que el procesador esté activo, utiliza el método sp.processor.stats() en mongosh:

sp.solarDemo.stats()

Este método informa las estadísticas operativas del procesador de flujo solarDemo.

También puedes utilizar el método sp.processor.sample() en mongosh para devolver una muestra de documentos procesados en el terminal.

sp.solarDemo.sample()

{
  _id: 10,
  max_temp: 16,
  avg_watts: 232,
  max_watts: 414,
  min_watts: 73
}

En la lista de procesadores de Stream Processing para tu espacio de trabajo de Stream Processing, haz clic en el icono Delete () de tu procesador de Stream Processing.

En el cuadro de diálogo de confirmación que aparece, escriba el nombre del procesador de flujo (solarDemo) para confirmar que desea eliminarlo y luego haga clic en Delete.

Utilice el método en sp.processor.drop() para mongosh solarDemoeliminar:

sp.solarDemo.drop()

Para confirmar que haya descartado solarDemo, utilice el sp.listStreamProcessors() método para enumerar todos sus procesadores de flujo disponibles:

sp.listStreamProcessors()