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 fluxo depende de conexões com 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 o desempenho e a segurança ideais com o Atlas Stream Processing, considere as práticas descritas nesta seção.
Emparelhamento e tabelas de roteamento
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 VPC ou VNet. Essas conexões protegem seus dados da exposição à Internet pública. Observe que, além de estabelecer a conexão de emparelhamento, você deve configurar explicitamente as tabelas de roteamento para a VPC do aplicativo para direcionar o tráfego para o bloco CIDR da Atlas VPC. Para saber mais sobre como configurar sua tabela de roteamento, 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 conexões entre projetos. As conexões entre projetos transmitem dados sem expor clusters à Internet privada, o que garante a privacidade sem a necessidade de configurar manualmente o emparelhamento VPC.
Autenticação AWS sem senha
O Atlas Stream Processing é compatível com as seguintes integrações da AWS:
S3
Streams de dados do Kinesis
Lambda (como
$externalFunction)
Cada uma dessas integrações é compatível com o Acesso AWS Unificado. A configuração do Acesso Unificado elimina a necessidade de armazenar IDs de acesso estático ou credenciais secretas no MongoDB e é o modelo de autenticação recomendado para todas as integrações da AWS.
Kafka OIDC
O Atlas Stream Processing oferece suporte à autenticação para agentes Apache Kafka usando o OIDC com JSON web tokens. No entanto, esse recurso atualmente é compatível apenas com fornecedores de identidade acessíveis em redes públicas. Esse recurso não é compatível com provedores de identidade na VPC de um cliente.
Tolerância a falhas
O Atlas Stream Processing fornece vários mecanismos para garantir um serviço confiável e robusto em cenários de falha comuns, incluindo checkpoints, dead letter queues (DLQ) e processadores de failover.
O Atlas Stream Processing monitora a integridade do processador de stream usando uma pulsação de 30segundos. Se um processador de stream não enviar uma pulsação por mais de 30 segundos, o Atlas Stream Processing reiniciará automaticamente o processador.
O Atlas Stream Processing gerencia automaticamente as alterações de topologia para integrações fragmentadas, como o AWS Kinesis.
Filas de letras mortas
Você pode configurar o Atlas Stream Processing para rotear dados não processados para uma fila de mensagens não entregues (DLQ) (DLQ). Em seguida, você pode visualizar registros não processados no DLQ para corrigir falhas de processamento.
Ao examinar seu DLQ, procure os seguintes erros comuns de processamento:
Análise de carga útil
O registro recebido está malformado e não é analisado. Isso geralmente resulta de JSON inválido.
Avaliação de expressão
Uma expressão no pipeline de processamento não consegue avaliar corretamente.
Exemplo
Seu pipeline inclui o campo de configuração
$emit.config.tombstoneWhenpara excluir registros Kafka e um determinado documento não tem o campo a ser avaliado.Limite de tamanho
O registro de saída excede o limite de tamanho do coletor alvo, como
1 MBpara o AWS Kinesis.Geração de chaves
O processador encontra um erro ao gerar uma chave da partição para o coletor.
Dados atrasados
Seu processador usa janelas e um determinado documento chegou após o período de atraso permitido configurado expirar.
Depois de analisar as causas das falhas de processamento, modifique suas fontes de dados upstream para corrigir os dados antes que eles entrem no pipeline do Atlas Stream Processing ou modifique seu pipeline para lidar com casos extremos.
Considere as seguintes opções de correção para registros no DLQ depois de usá-los para orientar as correções em sua configuração de Atlas Stream Processing :
Descartar
Se um registro for altamente malformado ou irrelevante – como no caso de dados muito atrasados em sistemas altamente sensíveis ao tempo – exclua-o da coleção.
Correção manual
Para um registro com erros JSON menores ou uma que seja crítica, ajuste manualmente os campos e insira-os diretamente no coletor de destino.
Reprocessamento automatizado:
Para grandes volumes de dados relevantes e recuperáveis, crie um script secundário ou um processador de fluxo adicional que leia do DLQ e aplique uma transformação em 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 em evento de falha do sistema em toda a região.
Princípios de configuração de failover
Os princípios a seguir definem uma configuração de failover robusta para proteger seus volumes de trabalho de Atlas Stream Processing .
Implemente clusters do Atlas em várias regiões.
Para garantir a operação contínua de seus processadores de stream em um evento de failover regional, qualquer cluster Atlas ao qual seu processador de stream se conecte deve ser um cluster multirregional. Esse princípio garante que suas conexões de origem ou coletor permaneçam ativas quando a região primária estiver inativa.
Crie espaços de trabalho do Atlas Stream Processing para cada região.
Após implantar seus clusters multirregional , crie um espaço de trabalho para cada região. Esse princípio minimiza a latência das conexões entre seus processadores de stream e os bancos de dados Atlas de suporte.
Preencha o Registro de conexão com conexões de backup.
Configure conexões para cada origem e coletor no espaço de trabalho de cada região. Certifique-se de configurar conexões não apenas para conexões básicas, mas para conexões de VPC e endpoints privados.
Configurar recuperação de desastres específicas da conexão
Para garantir a continuação do serviço para um processador de fluxo, quaisquer provedores externos nos quais ele confie também devem ter o failover habilitado. Certifique-se de que cada origem ou coletor de provedor externo esteja configurado corretamente de acordo com sua documentação.
Exemplo
Se o processador de fluxo usar um cluster Kafka como origem ou coletor, configure o cluster para preservação do deslocamento do consumidor para permitir que o Atlas Stream Processing retome o processamento a partir de um deslocamento do grupo de consumidores.
Realize execuções secas de rotina.
Verifique sua configuração de failover realizando periodicamente execuções secas de failover. Essas execuções secas servem para garantir que sua configuração se conecte às redes e carregue os pontos de verificação corretamente e para treinar o fluxo de trabalho de failover para que seja mais fácil executá-lo quando necessário. Considere estabelecer uma cadência regular de execuções secas de failover e execute uma execução seca toda vez que alterar sua configuração de failover.
Configurando o failover do processador de stream
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 a failover regional automatizado somente para processadores com as seguintes configurações:
Atlas source and sink
Origem Atlas e coletor Apache Kafka
Para obter mais informações sobre failover automatizado, consulte Processador de failover.
No entanto, você pode configurar uma arquitetura do Atlas Stream Processing que ofereça suporte a failover manual para qualquer combinação de conexão de origem e coletor.
Iniciar failover forçado
Você pode iniciar um failover forçado no nível do espaço de trabalho ou do processador de fluxo.
Para iniciar o failover do espaço de trabalho de processamento de fluxo, siga o procedimento descrito em Acionar failover do espaço de trabalho de processamento de fluxo.
Para iniciar o failover do processador de stream individual, siga o procedimento descrito em Iniciar failover para um processador de stream.
Respondendo a um evento de failover
No caso de um evento de interrupção de serviço regional:
Verifique a interrupção do serviço .
Ao notar sinais de interrupção de serviço , consulte o relatório de status de sua preferência para verificar o evento. Você pode visualizar essas informações na página de status da nuvem do MongoDB ou na página de status do seu fornecedor de nuvem. Você pode configurar Alertas de Atlas Stream Processing. Observe o timestamp do evento de falha .
Iniciar processadores de failover.
Siga o procedimento descrito em Iniciar failover para um processador de stream. Para garantir que o serviço seja retomado a partir do último evento de stream consumido, passe o timestamp que você anotou anteriormente como o parâmetro startAtOperationTime.
Os processadores que usam fontes Apache Kafka são retomados de acordo com as compensações do grupo de consumidores.
(Opcional) Automatizar a supressão do processador primário.
Quando sua região primary ficar on-line novamente, seus processadores primary 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 on-line.
Verifique se o serviço é retomado.
Depois de concluir o failover, verifique periodicamente o status da região primária do seu workspace. Você pode visualizar essas informações na página de status da nuvem do MongoDB ou na página de status do seu fornecedor de nuvem. Você também pode configurar Alertas de Atlas Stream Processing. Se você usar a supressão automática do processador primário, poderá configurar o script para retornar notificações quando ele detectar a reinicialização automática dos processadores primários.
Reinicie os processadores primários.
Reinicie os processadores de stream na sua região primária. Para garantir que o serviço seja retomado do último evento de stream consumido, passe o timestamp que você anotou anteriormente como o parâmetro startAtOperationTime.
Os processadores que usam fontes Apache Kafka são retomados de acordo com as compensações do grupo de consumidores.
Comportamento de failover
Considere o seguinte ao configurar seus sistemas para failover regional ou acionamento de failover manual:
Quando ocorre failover regional, o Atlas Stream Processing realiza os processadores de failover inativos em sua região de failover para processadores ativos. Esses processadores retomam o processamento de fluxo a partir do último checkpoint disponível .
Os checkpoints do Atlas Stream Processing têm a garantia de não ter mais de 10 minutos. No entanto, os processadores de failover reprocessam todos os dados que o processador primário processou desde o último checkpoint na inicialização. O failover garante a semântica de processamento pelo menos uma vez.
Se um processador de failover não iniciar durante um evento de failover regional, ele ainda assumirá 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
O Atlas Stream Processing fornece várias ferramentas de observabilidade para avaliar o desempenho e a integridade de seus espaços de trabalho, conexões e processadores. Além disso, para acompanhar o status e as interrupções do serviço regional, você pode acessar a página Status da nuvem do MongoDB .
Dimensionar o Atlas Stream Processing
O Atlas Stream Processing provisiona recursos por processador de stream. Para saber mais, consulte o Guia de seleção de nível.
Para processadores Atlas Stream Processing que operam em Atlas clusters, certifique-se de também dimensionar seus processadores e clusters proporcionais às necessidades de E/S do pipeline. Em particular, avalie os processadores que executam um alto volume de operações $lookup ou $merge como candidatos ao upscaling de cluster.
Considerações sobre o change stream
O Atlas Stream Processing pode usar um Atlas cluster $source para consumir eventos de change stream. Para garantir o desempenho e a confiabilidade ideais para esses processadores de stream, revise as seguintes considerações:
Tamanho da janela de oplog
Os processadores de stream aproveitam o oplog do cluster
$sourcepara retomar o serviço após a interrupção. Se um processador estiver inativo por um período maior que o período de retenção do oplog, o token de retomada poderá ponto para dados truncados, causando um erro deChangeStreamHistoryLost.Mitigue esse risco alinhando o dimensionamento do oplog com o tempo de inatividade aceitável para seus sistemas de processamento de fluxo.
Opções de retomada
O Atlas Stream Processing permite retomar o change stream
$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 em torno das opções de retomada que você usa em um determinado cenário.Ingestão de backlog
Os processadores de stream que operam em um namespace grande podem passar por longos períodos iniciais de processamento de recuperação. Considere reduzir o escopo do namespace do seu processador de stream ou escolher um ponto de início mais recente para reduzir esse período de recuperação inicial.
Alterar imagens de fluxo
Auditoria, transações de compensação e operações de desfazer exigem acesso ao estado de pré-processamento do documento . Ao trabalhar com o Atlas como fonte,habilite Document Pre e Post-Images para dar suporte a essas operações. Ao implementar o suporte para pré ou pós-imagens de documento , verifique a cobertura e a estabilidade em todo o sistema.
O estágio ``$replaceRoot``
As fontes de change stream retornam documentos que agrupam metadados operacionais com o conteúdo do documento afetado . A gravação desses documentos em um coletor exige que os consumidores trabalhem com o esquema de eventos de alteração em vez da lógica comercial do documento de origem. A menos que você precise dos metadados, use o estágio
$replaceRootpara promover o subdocumento do conteúdo de origem antes do processamento posterior.Pipelines orientados a união
O Atlas Stream Processing oferece suporte ao estágio
$lookuppara permitir enriquecimentos de dados de streaming. Essas operações podem aumentar a latência de processamento. Considere limitar os pipelines do Atlas Stream Processing apenas às operações$lookupque devem ser executadas antes de gravar em um coletor, reservando enriquecimentos mais complexos para visualizações pré-computadas.Checkpoints obsoletos
Os checkpoints contêm apenas o histórico de evento , não a lógica de processamento. Quando você retoma a operação a partir de um checkpoint, o processador de stream aplica a lógica do pipeline atual aos eventos antigos, o que pode causar uma incompatibilidade entre o estado downstream existente e a nova saída. Considere a introdução de um novo processador de fluxo em vez de retomar a partir de um checkpoint se você exigir uma semântica de processamento significativamente diferente.
Reiniciar o planejamento
Durante a reinicialização, um processador de fluxo pode apresentar uma taxa de transferência reduzida de novos eventos. Para aumentar a resiliência do sistema, o agendamento é reiniciado para levar em conta a capacidade temporariamente reduzida.
Planejamento de failover
Desenvolva procedimentos padrão para compartilhar os pontos de verificação do processador de fluxo entre regiões, escolhendo pontos de retomada seguros e validando a consistência da saída para garantir que eventos de failover regional não levem a problemas de qualidade de dados.