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:

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

1. Conéctese a su espacio de trabajo de procesamiento de flujo.

   Utilice la cadena de conexión asociada con su espacio de trabajo de procesamiento de flujo para conectarse mongosh usando.

   1. En el panel de su espacio de trabajo de procesamiento de flujo, haga 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 se muestra en el cuadro de diálogo. Tiene el siguiente formato:<atlas-stream-processing-url> es la URL de su espacio de trabajo de procesamiento de flujos y <username> es el nombre de usuario de la base de datos con el atlasAdmin rol.

      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.

      Presione Enter para ejecutarlo y conectarse a su espacio de trabajo de procesamiento de transmisión.

2. Configurar una $source etapa.

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

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

3. Configurar una $group etapa.

   Defina una variable para una etapa $group que derive la temperatura máxima y los vatajes promedio, máximo y mínimo de cada dispositivo solar de acuerdo con 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. Configurar una $tumblingWindow etapa.

   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 cuando la etapa $group calcula un valor para max_watts, extrae el valor máximo de los valores obs.watts para todos los documentos con un group_id determinado ingerido en los 10 segundos anteriores.

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

5. Configurar una etapa $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 consulta definida previamente y escribe los datos procesados ​​en la colección solarColl de la base de datos solarDb del clúster al que se conectó. Devuelve diversas mediciones derivadas de intervalos de 10segundos de observaciones de sus dispositivos solares.

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

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é escribiendo datos en su clúster 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 ir a la página Clusters y hacer clic en Data Explorer debajo del encabezado Shortcuts.

3. Ver la colección MySolar.

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

sp.solarDemo.stats()

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

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

sp.solarDemo.sample()

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

En la lista de procesadores de flujo de su espacio de trabajo de procesamiento de flujo, haga clic en el Delete ícono ( ) para su procesador de flujo.

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 ha solarDemo descartado, utilice el método para enumerar todos los procesadores de flujo sp.listStreamProcessors() disponibles:

sp.listStreamProcessors()