Introdução ao Atlas Stream Processing

Para criar um processador de fluxo na IU do Atlas, acesse a página Stream Processing do seu projeto do Atlas e clique em Configure no painel do seu espaço de trabalho do processamento de fluxos. Em seguida, escolha entre usar o construtor visual ou o editor JSON para configurar um processador de fluxo chamado solarDemo:

Execute os seguintes comandos em mongosh para criar um processador de fluxo persistente denominado solarDemo:

1. Conecte-se ao seu espaço de trabalho do processamento de fluxos.

   Use a string de conexão associada ao seu espaço de trabalho do processamento de fluxos para conectar usando mongosh.

   1. No painel da sua área de trabalho do processamento de fluxos, clique em Connect.

   2. Na caixa de diálogo Connect to your workspace, selecione a aba Shell.

   3. Copie a string de conexão exibida na caixa de diálogo. Ele tem o seguinte formato, onde <atlas-stream-processing-url> é a URL do seu espaço de trabalho de processamento de fluxo e <username> é o nome de usuário de um usuário de banco de dados com o atlasAdmin papel:

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

   4. Cole a string de conexão no seu terminal e substitua o espaço reservado <password> pelas credenciais do usuário.

      Pressione "Enter" para executar e conectar-se ao seu espaço de trabalho do processamento de fluxos.

2. Configure um estágio $source.

   Defina uma variável para um estágio $source que ingere dados da origem sample_stream_solar.

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

3. Configure um estágio $group.

   Defina uma variável para um estágio $group que deriva a temperatura máxima e as potências média, máxima e mínima de cada dispositivo solar de acordo com seu 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. Configure um estágio $tumblingWindow.

   Para realizar acumulações como $group em dados de streaming, o Atlas Stream Processing usa o Windows para vincular o conjunto de dados. Defina uma variável para um estágio $tumblingWindow que separa o fluxo em intervalos consecutivos de 10segundos.

   Isso significa, por exemplo, que quando o estágio $group calcula um valor para max_watts, ele extrai o valor máximo dos valores obs.watts para todos os documentos com um determinado group_id ingerido nos 10 segundos anteriores.

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

5. Configurar um estágio $merge .

   Defina uma variável para um estágio do $merge que escreve os dados de streaming processados em uma collection chamada solarColl no banco de dados do solarDb do seu Atlas cluster conectado.

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

6. Crie o processador de stream.

   Use o método sp.createStreamProcessor() para atribuir um nome ao seu novo processador de fluxo e declarar seu pipeline de agregação . O estágio $group pertence ao pipeline aninhado do $tumblingWindow, e você não deve incluí-lo na definição de pipeline do processador.

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

   Isso cria um processador de fluxo chamado solarDemo que aplica a query definida anteriormente e grava os dados processados na coleção solarColl do banco de dados solarDb no cluster ao qual você se conectou. Ele retorna várias medições derivadas de intervalos de 10 segundos de observações de seus dispositivos solares.

   Para saber mais sobre como o Atlas Stream Processing grava em bancos de dados at-rest, consulte $merge (Processamento de Stream).

Na lista de processadores do processamento de fluxos para sua área de trabalho do processamento de fluxos, clique no ícone Start para seu processador de fluxos.

Use o método sp.processor.start() em mongosh:

sp.solarDemo.start()

Para verificar se o processador de fluxo está gravando dados no cluster do Atlas:

1. No Atlas, acesse a página Clusters do seu projeto.

   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 o projeto desejado no menu Projects na barra de navegação.

   3. Na barra lateral, clique em Clusters sob o título Database.

      A página Clusters é exibida.

2. No Atlas, acesse a página Data Explorer do seu projeto.

   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 Data Explorer sob o título Database.

      O Data Explorer é exibido.

   Observação

   Você também pode ir para a página Clusters e clicar em Data Explorer sob o título Shortcuts.

3. Veja a coleção MySolar.

Para verificar se o processador está ativo, use o método sp.processor.stats() em mongosh:

sp.solarDemo.stats()

Este método relata estatísticas operacionais do processador de fluxo solarDemo.

Você também pode usar o método sp.processor.sample() em mongosh para retornar uma amostragem de documentos processados no terminal.

sp.solarDemo.sample()

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

Na lista de processadores de fluxo para seu espaço de trabalho do processamento de fluxos, clique no ícone Delete () do seu processador de fluxo.

Na caixa de diálogo de confirmação exibida, digite o nome do processador de stream (solarDemo) para confirmar que você deseja excluí-lo e clique em Delete.

Use o método sp.processor.drop() em mongosh para soltar solarDemo:

sp.solarDemo.drop()

Para confirmar que você descartou solarDemo, use o método sp.listStreamProcessors() para listar todos os seus processadores de stream disponíveis:

sp.listStreamProcessors()