Este guia descreve as melhores práticas para gerenciar o Atlas Stream Processing, incluindo procedimentos específicos para fluxos de trabalho essenciais.
Networking
Cada processador de stream depende de conexões para fontes de dados e coletores. O Atlas Stream Processing criptografa todos os dados em trânsito usando TLS/SSL, mas as conexões básicas ainda passam dados pela internet pública. Para obter desempenho e segurança ideais com o Atlas Stream Processing, considere as práticas descritas nesta seção.
Emparelhamento e Tabelas de rotas
Para garantir uma comunicação segura entre o Atlas Stream Processing e sistemas externos privados, como clusters Apache Kafka autogerenciados ou APIs privadas, use conexões de emparelhamento de VPC ou VNet. Essas conexões protegem seus dados contra a exposição à internet pública. Observe que, além de estabelecer a conexão de emparelhamento, você deve configurar explicitamente as tabelas de rotas para a VPC do seu aplicativo para direcionar o tráfego para o bloco CIDR da sua VPC do Atlas. Para aprender mais sobre como configurar sua tabela de rotas, consulte a documentação da VPC do seu provedor externo.
Conexões entre projetos do Atlas
Para rotear dados de streaming entre diferentes projetos do Atlas dentro da mesma organização, o Atlas Stream Processing oferece suporte a Cross Project Connections. As Conexões entre projetos transmitem dados sem expor os clusters à internet privada, o que garante a privacidade sem a necessidade de configurar manualmente o emparelhamento de VPC.
Autenticação AWS sem senha
O Atlas Stream Processing suporta as seguintes integrações AWS:
S3
Kinesis Data Streams
Lambda (como
$externalFunction)
Cada uma dessas integrações é compatível com Acesso unificado ao Amazon Web Services. A configuração do Unified Access elimina a necessidade de armazenar IDs de acesso estáticos ou credenciais Secret no MongoDB e é o modelo de autenticação recomendado para todas as integrações AWS.
Kafka OIDC
O Atlas Stream Processing oferece suporte à autenticação em brokers do Apache Kafka usando OIDC com JSON Web Tokens. No entanto, este recurso atualmente oferece suporte apenas a provedores de identidade que você pode acessar por redes públicas. Este recurso não é compatível com provedores de identidade dentro da VPC de um cliente.
Tolerância a falhas
Atlas Stream Processing provides various mechanisms to ensure reliable, robust service in common failure scenarios, including checkpoints, dead letter queues, and failover processors.
O Atlas Stream Processing monitora a saúde do processador de stream usando um pulso de 30segundos. Se um processador de fluxo falhar ao enviar um pulso de vida por mais de 30 segundos, o Atlas Stream Processing reinicia automaticamente o processador.
O Atlas Stream Processing gerencia automaticamente as alterações de topologia para integrações sharded como AWS Kinesis.
fila de mensagens não entregues (DLQ)
Você pode configurar o Atlas Stream Processing para rotear dados não processados para uma fila de mensagens não entregues (DLQ). Você pode então visualizar os registros não processados na DLQ para corrigir falhas em processamento.
The DLQ captures per-document processing failures only. Other errors prevent the processor from running and don't produce DLQ records. Such errors include invalid pipeline syntax or configuration and errors within the Atlas Stream Processing system. For these errors, Atlas Stream Processing applies its error handling and retry policies.
If a document fails processing and your stream processor has no configured DLQ, Atlas Stream Processing drops the document. DLQ statistics such as stats.dlqMessageCount still record the failure. If you sample the processor with sp.processor.sample(), the failed document still appears in the sample results.
Some pipeline stages define their own error handling behavior, which determines whether a failed document reaches the DLQ:
Estágio | Campo | Error Handling Behavior |
|---|---|---|
|
| |
|
| |
|
|
Ao examinar sua DLQ, procure pelos seguintes erros comuns em processamento:
Análise de payload
O registro de entrada está malformado e não pode ser analisado. Isso geralmente resulta de JSON inválido.
Avaliação de expressão
Uma expressão no pipeline em processamento falha ao ser avaliada corretamente.
Exemplo
Sua pipeline inclui o campo de configuração
$emit.config.tombstoneWhenpara excluir registros do Kafka e um determinado documento está faltando o campo a ser avaliado.Limite de tamanho
O registro de saída excede o limite de tamanho do seu destino de saída, como
1 MBpara Amazon Web Services Kinesis.Geração de chaves
O processador encontra um erro ao gerar uma chave da partição para o seu sink.
Dados Atrasados
Seu processador usa janelas e um determinado documento chegou após o período configurado de tolerância de atraso expirar.
Each DLQ record contains the failed document and details of the failure. For the full DLQ document schema, see the Monitoring guide.
Depois de analisar as causas das falhas em processamento, modifique suas fontes de dados upstream para corrigir os dados antes que eles entrem no seu pipeline do Atlas Stream Processing ou modifique seu pipeline para lidar com casos extremos.
Considere as seguintes opções de correção para registros na DLQ após usá-los para orientar as correções na sua configuração do Atlas Stream Processing:
Descartar
Se um registro estiver muito malformado ou for irrelevante—por exemplo, dados muito atrasados em sistemas com alta sensibilidade ao tempo—exclua-o da coleção.
Correção manual
Para um registro com pequenos erros de JSON ou um que seja de missão crítica, ajuste manualmente os campos e insira-os diretamente no seu sink de destino.
Reprocessamento automatizado:
Para grandes volumes de dados recuperáveis e relevantes, crie um script secundário ou um processador de fluxo adicional que leia da DLQ e aplicar uma transformação em relação a condições de falha conhecidas antes de rotear os dados limpos para o coletor de destino.
falha
O Atlas Stream Processing fornece processadores de failover para evitar a interrupção do serviço no caso de uma falha do sistema em toda a região.
Princípios de configuração de failover
Os seguintes princípios definem uma configuração de failover robusta para proteger suas cargas de trabalho do Atlas Stream Processing.
Implantar clusters Atlas em várias regiões.
Para garantir a operação contínua dos seus processadores de stream em um evento de failover regional, use clusters multirregionais para qualquer cluster Atlas ao qual seu processador de stream se conecta. O Atlas Stream Processing não bloqueia o failover para cluster de região única, mas esse princípio garante que suas conexões de origem ou destino permaneçam ativas quando a região primário estiver inativa.
Crie espaços de trabalho do Atlas Stream Processing para cada região.
Depois de implantar seu cluster multirregional, crie um workspace para cada região. Este princípio minimiza a latência das conexões entre seus processadores de fluxo e os bancos de dados Atlas que os suportam.
Popular o Registro de Conexões com conexões de backup.
Configure as conexões para cada origem e destino no workspace de cada região. Certifique-se de configurar conexões não apenas para conexões básicas, mas também para VPC e conexões de private endpoints.
Configurar recuperação de desastre específica da conexão
Para garantir a continuidade do serviço de um processador de fluxo, quaisquer provedores externos dos quais ele depende também devem ter o failover ativado. Certifique-se de que cada fonte ou coletor de provedor externo esteja configurado corretamente de acordo com sua documentação.
Exemplo
Se o seu processador de fluxo usar um cluster Kafka como origem ou destino, configure o cluster para preservação de offset do consumidor para permitir que o Atlas Stream Processing retome em processamento a partir de um offset de grupo de consumidores.
Realize simulações de rotina.
Verifique sua configuração de failover realizando periodicamente simulações de failover. Essas simulações servem tanto para garantir que sua configuração se conecte a redes e carregue checkpoints corretamente quanto para treinar o fluxo de trabalho de failover para que seja mais fácil de executar quando necessário. Considere estabelecer uma cadência regular de simulações de failover e execute uma simulação sempre que alterar sua configuração de failover.
Configurando o failover do Stream Processor
Depois de garantir que as melhores práticas estejam em vigor, crie um processador de fluxo e verifique sua configuração de failover.
O Atlas Stream Processing oferece suporte ao failover regional automatizado apenas para processadores com as seguintes configurações:
Atlas source and sink
Fonte do Atlas e Apache Kafka sink
Para obter mais informações sobre o failover automatizado, consulte Processador de Failover.
No entanto, você pode configurar uma arquitetura Atlas Stream Processing que oferece suporte a failover manual para qualquer combinação de conexão de origem e destino.
Iniciar failover forçado
Você pode iniciar um failover forçado no nível do workspace ou do processador de fluxo.
Para iniciar o failover do Stream Processing workspace, siga o procedimento descrito em trigger failover do Stream Processing workspace.
Para iniciar o failover de um processador de fluxo individual, siga o procedimento descrito em Iniciar failover para um processador de fluxo.
Respondendo a um evento de failover
Em caso de um evento de interrupção de serviço regional:
Verifique a interrupção do serviço.
When you notice symptoms of a service outage, review your preferred status report to verify the event. You can view this information on the MongoDB Cloud Status page or your cloud provider's status page. You can configure Atlas Stream Processing Alerts. Note the timestamp of the failure event.
Inicie os processadores de failover.
Siga o procedimento descrito em Iniciar failover para um Stream Processor. Para garantir que o serviço seja retomado a partir do último evento de stream consumido, passe o carimbo de data/hora que você anotou anteriormente como o parâmetro startAtOperationTime.
Processadores que usam fontes do Apache Kafka retomam de acordo com os offsets do grupo de consumidores.
(Opcional) Automatize a supressão do processador primário.
Quando sua região primária voltar a ficar online, seus processadores primários serão reiniciados automaticamente, resultando em processamento duplicado. Crie e execute um script para verificar continuamente o status da região primária e interromper todos os processadores nessa região quando ela voltar a ficar online.
Verifique se o serviço é retomado.
After completing failover, periodically check the status of your workspace's primary region. You can view this information on the MongoDB Cloud Status page or your cloud provider's status page. You can also configure Atlas Stream Processing Alerts. If you use automatic primary processor suppression, you can configure the script to return notifications when it detects primary processors automatically restarting.
Reinicie os processadores primários.
Reinicie os processadores de fluxo na sua região primária. Para garantir que o serviço seja retomado a partir do último evento de stream consumido, passe o carimbo de data/hora que você anotou anteriormente como o parâmetro startAtOperationTime.
Processadores que usam fontes do Apache Kafka retomam de acordo com os offsets do grupo de consumidores.
Comportamento de failover
Considere o seguinte ao configurar seus sistemas para failover regional ou acionar failover manual:
Quando ocorre um failover regional, o Atlas Stream Processing promove os processadores de failover inativos na sua região de failover para processadores ativos. Esses processadores retomam o Stream Processing a partir do último checkpoint disponível.
Os checkpoints do Atlas Stream Processing têm garantias de não ter mais de 10 minutos. No entanto, os processadores de failover reprocessam todos os dados que seu processador primário processou desde o último checkpoint na inicialização. O failover oferece garantias de semânticas em processamento de pelo menos uma vez.
Se um processador de failover não iniciar durante um evento de failover regional, ele ainda assume a função de processador ativo. Você pode então solucionar a causa subjacente e reiniciar o processador normalmente na região de failover.
Observabilidade
Atlas Stream Processing provides various observability tools to assess the performance and health of your workspaces, connections, and processors. Additionally, to track regional service status and outages, you can visit the MongoDB Cloud Status page.
Dimensionar o Atlas Stream Processing
O Atlas Stream Processing provisiona recursos por processador de fluxo. Para aprender mais, consulte o Tier de Seleção.
Para processadores do Atlas Stream Processing que operam em clusters do Atlas, certifique-se de também dimensionar seus processadores e clusters proporcionalmente às necessidades de E/S do seu pipeline. Em particular, avalie os processadores que realizam um alto volume de operações de $lookup ou $merge como candidatos para o aumento de escala do cluster.
Considerações sobre o change stream
O Atlas Stream Processing pode usar um cluster do Atlas $source para consumir eventos de fluxo de alterações. Para garantir o desempenho e a confiabilidade ideais para esses processadores de fluxo, revise as seguintes considerações:
Tamanho da Oplog Window
Stream processors leverage the
$sourcecluster's oplog to resume service after stoppage. If a processor is down for a period greater than the oplog retention period, the resume token can point to truncated data, causing aChangeStreamHistoryLosterror.Mitigue esse risco alinhando o dimensionamento do oplog com o tempo de inatividade aceitável para seus sistemas de Stream Processing.
Opções de Retomada
O Atlas Stream Processing oferece suporte para retomar o fluxo de alterações
$source``s from either the last stored checkpoint with ``resumeFromCheckpointou um horário específico comstartAtOperationTime. Para evitar lacunas ou duplicações nos dados processados, defina uma política padronizada para as opções de retomada que você usa em um determinado cenário.Ingestão de backlog
Processadores de fluxo que operam em um namespace grande podem experimentar longos períodos iniciais em processamento de recuperação. Considere reduzir o escopo do namespace do seu processador de fluxo ou escolher um ponto de partida mais recente para reduzir esse período inicial de sincronização.
Imagens de Change Stream
Auditing, compensating transactions, and undo operations require access to pre-processing document state. When you work with Atlas as a source, enable Document Pre- and Post-Images to support these operations. When you roll out support for document pre- or post-images, verify coverage and stability across your system.
O Estágio ``$replaceRoot``
As fontes de fluxo de alteração retornam documentos que agrupam metadados operacionais com o conteúdo do documento afetado. Gravar esses documentos em um destino exige que os consumidores trabalhem com o esquema de evento de alteração em vez da lógica de negócios do documento de origem. A menos que você precise dos metadados, use o estágio
$replaceRootpara promover o subdocumento de conteúdo de origem antes de em processamento adicional.pipeline orientada a junção
O Atlas Stream Processing suporta o estágio
$lookuppara permitir enriquecimentos semelhantes a junções de dados de streaming. Essas operações podem aumentar a latência em processamento. Considere limitar seus pipelines do Atlas Stream Processing a apenas aquelas$lookupoperações que devem ser realizadas antes de gravar em um sink, reservando enriquecimentos mais complexos para visualizações pré-calculadas.Checkpoints obsoletos
Os checkpoints contêm apenas o histórico de eventos, não a lógica em processamento. Quando você retoma a operação de um checkpoint, seu processador de fluxo aplica a lógica do seu pipeline atual aos eventos antigos, o que pode causar uma incompatibilidade entre seu estado downstream existente e sua nova saída. Considere introduzir um novo processador de fluxo em vez de retomar de um checkpoint se você precisar de uma semântica em processamento significativamente diferente.
Planejamento de Reinício
Durante a reinicialização, um processador de fluxo pode experimentar uma taxa de transferência reduzida de novos eventos. Para aumentar a resiliência do sistema, agende reinícios para lidar com a capacidade temporariamente reduzida.
Planejamento de failover
Desenvolva procedimentos padrões para o compartilhamento de checkpoints de processadores de fluxo entre regiões, escolhendo pontos de retomada seguros e validando a consistência da saída para garantir que os eventos de failover regional não levem a problemas de qualidade dos dados.