Dimensione corretamente seus streams com estado mais exigentes, desde a detecção de fraudes até o monitoramento em tempo real, sem nunca precisar recriar o estado do seu checkpoint
por Thangam Vaiyapuri, Jay Palaniappan, B. Micheal Okutubo e Zifei Feng
Qualquer pessoa que execute consultas stateful do Apache Spark™ Structured Streaming em produção acaba se deparando com a mesma barreira incômoda.
Você iniciou a consulta há meses. Naquela época, o volume de dados era modesto, então você aceitou o padrão de 200 partições de shuffle e seguiu em frente. O pipeline funcionou perfeitamente. Depois, a empresa cresceu, o tráfego triplicou e o armazenamento de estado inflou. De repente, essas 200 partições não têm mais o tamanho adequado. Algumas partições estão desbalanceadas e sobrecarregadas, o cluster está sob pressão e cada microbatch demora mais do que deveria.
Então você faz o que é natural: aumenta o spark.sql.shuffle.partitions e reinicia a consulta. Nada muda.
A consulta ignora silenciosamente o seu novo valor porque a contagem de partições foi gravada no checkpoint quando você iniciou o stream pela primeira vez. Historicamente, a única maneira de aplicar um novo número era abandonar o checkpoint existente e recomeçar do zero, o que, para uma consulta stateful, significa perder todo o estado acumulado que você manteve cuidadosamente. Para um modelo de fraude que rastreia milhões de contas ou um job de sessionization que mantém dias de janelas, "recomeçar do zero" não é uma frase que alguém queira dizer em uma revisão de incidentes de produção.
O reparticionamento de estado sob demanda (Public Preview), disponível no Databricks Runtime 18 e superior, elimina essa barreira. Agora você pode redimensionar o número de partições para uma consulta de streaming stateful e manter o estado do seu checkpoint intacto.
Isso se aplica a qualquer consulta de streaming stateful, quer você execute agregações, junções stream-stream, deduplicação, sessionization ou transformWithState, e a qualquer carga de trabalho, desde detecção de fraudes até monitoramento em tempo real.
Para os primeiros usuários como a Coveo, a capacidade de dimensionar corretamente sua infraestrutura de streaming sob demanda se traduziu imediatamente em economias operacionais significativas.
Na Coveo, executamos pipelines de streaming stateful em grande escala, onde os volumes de dados flutuam significativamente ao longo do tempo. Com o Databricks e o recurso de reparticionamento de estado, reduzimos nossos custos relacionados à API do Amazon S3 em 40%. Antes, cada decisão de dimensionamento forçava uma escolha difícil: superprovisionar ou reconstruir a partir de novos checkpoints, o que elevava os custos de API de armazenamento a quase o mesmo nível dos custos de computação. Agora, dimensionamos livremente conforme a demanda muda, sem interromper o estado existente ou acionar migrações de checkpoint dispendiosas.” —Alexis Chicoine, Desenvolvedor de Software Sênior, Coveo
Para entender por que os resultados da Coveo representam um avanço significativo para o Structured Streaming, precisamos analisar por que a contagem de partições era congelada em primeiro lugar.
Uma consulta de streaming stateful mantém seu estado em um armazenamento de estado, e esse estado é fisicamente particionado. Cada chave no seu stream — um ID de usuário, um número de conta, uma janela — é mapeada (hashed) para uma partição específica, e os dados de cada partição são armazenados em sua própria instância separada do RocksDB dentro do checkpoint. O número de partições define o layout de todo o armazenamento de estado no disco.
Se você simplesmente alterasse a contagem de partições entre as reinicializações, o mapeamento (hashing) não coincidiria mais. Uma chave que antes residia em uma partição (digamos, partição 47) agora poderia ser mapeada para uma diferente (partição 12), mas seu estado acumulado ainda estaria nos arquivos da partição original. A consulta, na prática, perderia o controle de sua própria memória. Para evitar exatamente esse tipo de corrupção silenciosa, o Structured Streaming bloqueava a contagem de partições na criação do checkpoint e ignorava quaisquer alterações posteriores em spark.sql.shuffle.partitions.
Seguro, mas inflexível. Os dois preços que você pagava eram:
O reparticionamento de estado sob demanda resolve ambos os problemas fazendo a única coisa que o design antigo se recusava a fazer — mas de forma segura, redistribuindo fisicamente o estado para corresponder à nova contagem de partições.
Os requisitos são simples:
Essa é a lista completa de pré-requisitos. Se você estiver no DBR 18 com o armazenamento de estado padrão, já tem tudo o que precisa.
O mecanismo é simples e reutiliza um padrão que todo desenvolvedor de streaming já conhece: parar, reconfigurar, reiniciar.
Em vez de spark.sql.shuffle.partitions, você define uma configuração dedicada, spark.sql.streaming.stateStore.partitions, e reinicia a consulta:
O detalhe principal é a própria nova configuração. Para consultas stateful, spark.sql.streaming.stateStore.partitions tem precedência sobre spark.sql.shuffle.partitions. É isso que faz com que a alteração persista onde a abordagem antiga falhava.
Quando a consulta é reiniciada, ela não retoma o processamento normal imediatamente. Primeiro, ela conclui o último microbatch planejado, se houver algum pendente. Em seguida, ela realiza uma operação única de reparticionamento: redistribui fisicamente os dados de estado entre o novo número de partições, mapeando novamente as chaves para seus novos locais corretos para que nada seja perdido ou extraviado. Assim que essa redistribuição for concluída, a consulta retoma o processamento normal, agora usando a contagem de partições que você solicitou.
Essa etapa de reparticionamento é o coração do recurso. É a diferença entre "alteramos um número" e "movemos seu estado com segurança para um novo layout".
Como o reparticionamento é uma operação real cujo tempo de execução é proporcional à quantidade de estado, você vai querer ter visibilidade sobre ele. O Structured Streaming expõe isso por meio de seus relatórios de progresso padrão.
Após a conclusão do próximo microbatch, os eventos StreamingQueryProgress incluirão a duração da operação de reparticionamento. Procure nas métricas durationMs do evento pelo campo controlBatch.REPARTITION, que informa a duração do reparticionamento em milissegundos.
Um volume de estado maior significa um reparticionamento mais longo, mas esperamos que leve apenas alguns segundos para a maioria das cargas de trabalho. Portanto, em jobs grandes, vale a pena capturar essa métrica para entender a duração. Para saber mais sobre como ler esses eventos, consulte Monitorando consultas do Structured Streaming no Databricks.
Vamos tornar isso concreto com uma agregação simples: uma contagem de eventos por id com janela de tumbling. Começaremos com o padrão de 200 partições, decidiremos que isso é mais do que essa carga de trabalho precisa e reduziremos para 100.
Primeiro, a consulta como ela é executada hoje, com a contagem de partições padrão:
Agora, acompanhamos esse stream por um tempo e concluímos que 200 partições é um exagero. Estamos pagando uma sobrecarga de coordenação por um paralelismo de que não precisamos. Paramos a consulta, definimos a nova contagem de partições e a reiniciamos com as mesmas opções e o mesmo checkpoint:
Quando a consulta reiniciada entra em execução, ela finaliza o último microbatch planejado, se houver algum pendente, executa o reparticionamento para redistribuir o estado de 200 partições para 100 e, em seguida, continua a contagem com cada janela e cada total acumulado totalmente preservados. O mesmo procedimento funciona no sentido inverso: para aumentar a escala sob uma carga mais pesada, basta definir um número maior.
A mesma abordagem se aplica ao Spark Declarative Pipelines (SDP). Consulte o exemplo de SDP na documentação para obter um passo a passo completo.
O reparticionamento de estado sob demanda é uma ferramenta de ajuste e dimensionamento, e não uma operação rotineira. Ele se mostra valioso em algumas situações importantes:
Como cada alteração exige uma parada e reinicialização com uma pausa única de reparticionamento, trate isso como uma ação de manutenção deliberada. Planeje o redimensionamento para uma janela em que uma breve pausa no processamento seja aceitável, acompanhe controlBatch.REPARTITION para confirmar quanto tempo levou e deixe a consulta voltar ao seu ritmo normal.
Durante anos, a contagem de partições de uma consulta de streaming com estado era uma decisão tomada uma única vez, logo no início, e nunca mais revisitada — ou você pagava um preço alto para reconstruir o estado do zero. O reparticionamento de estado sob demanda elimina essas restrições. Redistribuir o estado com segurança em uma nova contagem de partições transforma uma decisão que antes era tomada apenas no início em algo que você pode revisitar sempre que sua carga de trabalho exigir.
O resultado é exatamente o que os operadores de streams de longa execução sempre quiseram: a liberdade de dimensionar corretamente uma consulta com base em suas necessidades de escala, com nada mais do que uma parada, uma alteração de configuração e uma reinicialização, sem perder seu estado.
O reparticionamento de estado sob demanda está disponível no Databricks Runtime 18 e superior, usando o provedor de armazenamento de estado RocksDB. Para obter a referência completa, consulte Reparticionamento de estado sob demanda para consultas de streaming com estado.
(Esta publicação no blog foi traduzida utilizando ferramentas baseadas em inteligência artificial) Publicação original
Assine nosso blog e receba os posts mais recentes diretamente na sua caixa de entrada.