Comenzar con Atlas Stream Processing

Para crear un procesador de Stream Processing 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 Stream Processing. Luego elige entre usar el editor visual o el editor JSON para configurar un procesador de flujos 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, selecciona 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. Pega la cadena de conexión en tu terminal y reemplaza el marcador <password> por 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 transmisión, Atlas Stream Processing utiliza ventanas para delimitar el conjunto de datos. Define una variable para una etapa $tumblingWindow que separe la transmisión en intervalos consecutivos de 10 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. Crea el procesador de flujos.

   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.

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

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 puedes hacer clic en el nombre de un clúster para abrir la barra lateral Cluster, y luego hacer clic en Data Explorer bajo el encabezado Shortcuts.

3. Mira 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 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 diálogo de confirmación que aparece, escribe el nombre del procesador de flujo (solarDemo) para confirmar que deseas borrarlo y luego haz clic en Delete.

Usa el método sp.processor.drop() en mongosh para descartar solarDemo:

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()