Introdução ao desenvolvimento do Atlas Stream Processing
Avalie esse Início rápido
Bem-vindo a este tutorial de Atlas Stream Processing! Neste guia, configuraremos rapidamente um fluxo de trabalho de codificação e faremos com que você escreva e execute sua primeira instância de Atlas Stream Processing rapidamente. Em muito pouco tempo, aprenderemos como criar uma nova instância de processador de fluxo, codificar e executar processadores de fluxo de maneira conveniente a partir do Visual Studio Code e simplesmente agregar dados de fluxo, abrindo assim a porta para um campo totalmente novo do MongoDB Atlas. plataforma de dados.
- Pré-requisitos
- Configurar
- Criar uma instância do processador de fluxo
- Configurar o Visual Studio Code
- A Anatomia de um Processador de Stream
- Vamos executar um processador de fluxo!
- Dados codificados na declaração $source
- Processador de fluxo mais simples
- Agregação de processamento de fluxo
- Adicionar carimbos de data/hora aos dados
- Um MongoDB Atlas cluster ativo que oferece suporte ao Atlas Stream Processing
- Visual Studio Code + Extensão do MongoDB para VS Code
Precisamos ter uma Atlas Stream Processing Instance (SPI) pronta. Siga as etapas do tutorial Introdução ao Atlas Stream Processing: criando seu primeiro processador de streams até que tenhamos nossa connection string e nome de usuário/senha. Em seguida, volte aqui.
Não se trata de adicionar seu endereço IP ao Atlas Network Access para permitir que o cliente acesse a instância.
Graças à extensão MongoDB para VS Code, podemos desenvolver rapidamente pipelines de agregação de processamento de fluxo (SP) e executá-los diretamente de dentro de um playground do VS Code MongoDB. Isso proporciona uma experiência muito melhor para o desenvolvedor. No restante deste artigo, usaremos o VS Code.
Esse playground é um ambiente NodeJS onde podemos executar JavaScript interagindo com um processador de transmissão ao vivo no MongoDB Atlas. Para começar, instale oVS Code e a MongoDB for VS Code.
Abaixo está um ótimo tutorial sobre a instalação da extensão. Ele também lista alguns comandos de shell de que precisaremos mais tarde.
- Objetivo: Se tudo funcionar, devemos ver nossa conexão SP ativa na guia Extensão MongoDB.
Em uma instância de processamento de fluxo (=cluster), é possível criar e executar vários processadores de fluxo (=pipelines de processamento de fluxo) simultaneamente.
Um processador de fluxo único (SP) é muito semelhante a um pipeline de agregação do MongoDB. Ele é descrito por uma matriz de estágios de processamento. Entretanto, há algumas diferenças. O SP mais básico pode ser criado usando apenas sua fonte de dados (teremos exemplos executáveis a seguir).
Um processador de fluxo mais realista conteria pelo menos um estágio de agregação, e pode haver um grande número de estágios executando várias operações no fluxo de dados de entrada. Há um limite gratuito de 16MB para o tamanho total do processador.
Para aumentar a velocidade do loop de desenvolvimento, há uma função sp.process() que inicia um processador de fluxo efêmero que não persistirá em sua instância de processamento de fluxo.
Vamos criar processadores de fluxo básicos e construir nosso caminho. Primeiro, precisamos ter alguns dados! O Atlas Stream Processing oferece suporte a várias fontes de dados para eventos de streaming recebidos. Essas fontes incluem:
- Declaração de dados codificados em $source.
- Fluxos do Kafka.
- MongoDB Atlas databases.
Para testes rápidos ou exemplos independentes, ter um pequeno conjunto de dados codificados é uma maneira muito conveniente de produzir eventos. Podemos declarar uma matriz de eventos. Aqui está um exemplo extremamente simples e observe que faremos alguns ajustes posteriormente para cobrir diferentes casos de uso.
No VS Code, executamos um processador de fluxo efêmero com sp.process(). Dessa forma, não precisamos usarsp.createStreamProcessor() esp..drop() constantemente, como faria com os SPs que deveriam ser salvos permanentemente na instância.
Ao executar este playground, devemos ver os dados aparecendo na aba "OUTPUT" do VS Code (CTRL+SHIFT+U para fazer com que apareçam)
Observação: pode levar alguns segundos para que o SP seja carregado e executado, portanto, não espere uma saída imediata.
Este SP simples pode ser usado para garantir que os dados cheguem ao SP e que não haja problemas up stream com nossa fonte. Os dados de registros de data e hora foram gerados no momento da ingestão.
Com base no que temos, é fácil adicionar um pipeline de agregação simples ao nosso SP. Abaixo, estamos adicionando um estágio $group para agregar/acumular o campo " value " das mensagens recebidas em uma matriz para o intervalo solicitado.
Observe que o estágio "w" (w significa "Window") do pipeline SP contém um pipeline de agregação interno. Com o processamento de fluxo, temos pipelines de agregação no pipeline de processamento de fluxo.
Esse estágio apresenta um $tumblingWindow que define o período de tempo em que a agregação será executada. Lembre-se de que os fluxos devem ser contínuos, portanto, uma janela é semelhante a um buffer.
intervalo define a duração de tempo de uma janela. Como a janela é um fluxo contínuo de dados, só podemos agregar em uma fatia de cada vez.
idleTimeout define por quanto tempo o $source pode permanecer ocioso antes de fechar a janela. Isso é útil se o fluxo não for sustentado.
Deixe-o rodar por alguns segundos e devemos obter uma saída semelhante à seguinte. $group criará um documento por campo " id " de entrada e agregará os valores relevantes em um novo campo de matriz, " values. "
Dependendo das configurações de $tumblingWindow, a agregação produzirá vários documentos que correspondam aos carimbos de data/hora. Por exemplo, essas configurações...
... produzirá o seguinte resultado de agregação:
Veja como os campos windowStartTimestamp e windowEndTimestamp mostram os intervalos de 10segundos conforme solicitado (14:51:30 a 14:51:40 etc.).
Atlas Stream Processing usa a API de query MongoDB. Você pode saber mais sobre a API de query do MongoDB com a documentação oficial da API de query , o curso interativo[grátis] e o tutorial.
Importante: os pipelines de agregação de processamento de fluxo não oferecem suporte a todas as operações de agregação de banco de dados e possuem operadores adicionais específicos para streaming, como $tblingWindow. Verifique a documentação oficial de aggregation do Stream Processing.
Mesmo quando codificamos os dados, há uma oportunidade de fornecer um registro de data e hora caso queiramos realizar operações de $sort e imitar melhor um caso de uso real. Isso seria o equivalente a um registro de data e hora do evento incorporado à mensagem.
Existem muitos outros tipos de carimbos de data/hora se usarmos um stream Kafka ao vivo (atribuído pelo produtor, do lado do servidor, tempo de ingestão e muito mais). Adicione um carimbo de data/hora às nossas mensagens e use a propriedade "timeField" do documento para torná-lo o carimbo de data/hora do fluxo oficial.
Neste ponto, temos tudo o que precisamos para testar novos pipelines e criar provas de conceito de uma forma conveniente e independente. Em um artigo subsequente, demonstraremos como se conectar a várias fontes de streaming.
No momento da publicação da publicação, o Atlas Stream Processing está em pré-visualização pública e há uma série de limitações conhecidas do Stream Processing que você deve estar ciente, como disponibilidade do centro de dados regional, conectividade com outros projetos do Atlas e privilégios de usuário.
Ao executar um processador de fluxo efêmero via sp.process(), muitos erros (problema de serialização de JSON, dados atrasados, divisão por zero, erros de $validate) que poderiam ter ido para uma fila de letras mortas (DLQ) são enviados para a saída padrão para ajudá-lo a depurar.
Para SPs criados com max.createStreamProcessor(), você terá que configurar seu DLQ manualmente. Consulte a documentação para isso. Na página de documentação "Gerenciar o processador de fluxo", procure por "Definir um DLQ".
Após mesclar dados em um banco de dados Atlas, é possível usar as ferramentas de construção de agregação de pipeline existentes no construtor da GUI do Atlas ou MongoDB Compass para criar e depurar pipelines. Como essas ferramentas são destinadas à API do banco de dados principal, lembre-se de que alguns operadores não são suportados pelos processadores de stream, e recursos de streaming, como janelas, não estão disponíveis no momento.
Com isso, você deve ter tudo o que precisa para colocar seu primeiro processador de stream em funcionamento. Em uma postagem futura, entraremos em mais detalhes na conexão a diferentes fontes de dados para seus processadores de stream.
Se você tiver alguma dúvida, compartilhe-a em nosso fórum da comunidade, encontre-se conosco durante os Grupos de usuários locais do MongoDB (MEGs) ou confira um de nossos eventoslocais do MongoDB .
- Atlas Stream Processing já está em pré-visualização pública ( 13 de fevereiro 2024)
Principais comentários nos fóruns
Ainda não há comentários sobre este artigo.