Introdução ao Atlas Stream Processing

Para criar um processador de fluxo na interface do usuário do Atlas , vá para a página Stream Processing do seu projeto Atlas e clique em Configure no painel da sua instância de processamento de fluxo. Em seguida, escolha entre utilizar o construtor visual ou o editor JSON para configurar um processador de fluxo denominado solarDemo:

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

1. Conecte-se à sua instância de processamento de fluxo.

   Use a connection string associada à sua instância de Atlas Stream Processing para se conectar usando mongosh.

   1. No painel da instância do Atlas Stream Processing , clique em Connect.

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

   3. Copie a string de conexão exibida na caixa de diálogo. Ele tem o seguinte formato, em que <atlas-stream-processing-url> é a URL da sua instância de processamento de fluxo e <username> é o nome de usuário de um usuário de banco de dados com a função atlasAdmin:

      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 executá-lo e conecte-se à instância de processamento de fluxo.

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 de fluxo para sua instância de processamento de fluxo, clique no ícone Start para seu processador de fluxo.

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.

   AVISO: Melhorias de navegação em andamento

   No momento, estamos lançando uma experiência de navegação nova e aprimorada. Se as etapas a seguir não corresponderem à sua visualização na UI do Atlas, consulte a documentação de visualização.

   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. Se ainda não estiver exibido, clique em Clusters na barra lateral.

      A página Clusters é exibida.

2. Clique no botão Browse Collections para o seu cluster.

   O Data Explorer é exibido.

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 sua instância de processamento de fluxo, clique no ícone Delete () para 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()