Considerações sobre a produção de transmissão estruturada
Execute cargas de trabalho de produção de transmissão estructurada como Jobs de LakeFlow agendados no Databricks. Veja Jobs do LakeFlow.
A Databricks recomenda que você sempre configure o seguinte:
- Remova o código desnecessário dos notebooks que retornariam resultados, como
displayecount. - Não execute cargas de trabalho de transmissão estructurada usando compute de uso geral. Sempre programe as transmissões como Jobs do LakeFlow usando o compute de Jobs.
- Programe Jobs do LakeFlow usando o
Continuousmodo. Isso se refere ao recurso de programação do Databricks Jobs, e não ao intervalo de trigger da transmissão estructurada. - Não ative Autoscale para compute em jobs de transmissão estructurada.
Algumas cargas de trabalho se beneficiam do seguinte:
- Configurar RocksDB armazenamento do estado em Databricks
- Ponto de verificação de estado assíncrono para consultas com estado
- Monitoramento de progresso assíncrono
A Databricks introduziu os LakeFlow Pipelines para reduzir as complexidades do gerenciamento da infraestrutura de produção para cargas de trabalho de Structured Streaming. A Databricks recomenda usar os LakeFlow Pipelines para novos pipelines de Structured Streaming. See Spark Declarative Pipelines.
O autoscale de compute tem limitações na redução do tamanho do cluster para cargas de trabalho de Structured Streaming. A Databricks recomenda usar Spark Declarative Pipelines no LakeFlow com autoscale aprimorado para cargas de trabalho de transmissão. Consulte Otimizar a utilização de cluster de LakeFlow Pipelines com autoscale.
:::nota computesem servidor
Em compute serverless , apenas Trigger.AvailableNow() e Trigger.Once() são suportados. Databricks recomenda Trigger.AvailableNow().
Para transmissão contínua em compute serverless , use o modo de pipeline acionado em vez do modo pipeline contínuo no modo contínuo.
Veja limitações de transmissão.
:::
Reduzir a latência para transmissão operacional
As cargas de trabalho de transmissão operacional ingerem, transformam e agem sobre os dados em tempo quase real. Exemplos comuns incluem detecção de fraudes, detecção de anomalias, personalização e monitoramento e alertas em tempo real, onde o processamento atrasado afeta diretamente os resultados de negócios. A baixa latência para essas cargas de trabalho normalmente significa dezenas a centenas de milissegundos, embora muitas equipes definam acordos de nível de serviço (SLAs) na faixa de segundos para considerar a variabilidade em percentis mais altos.
Para a menor latência de ponta a ponta, use o modo em tempo real, que atinge uma latência de ponta a ponta inferior a um segundo no limite e cerca de 300 milissegundos em casos comuns. Consulte Conceitos do modo em tempo real.
Quando o modo em tempo real não se adequa à sua carga de trabalho, as seguintes práticas recomendadas reduzem a latência para o Structured Streaming em microlotes:
- Modo de saída : use o modo de atualização onde seus operadores de query e sink o suportam. O modo de atualização emite linhas atualizadas após cada trigger e continua atualizando-as até que a marca d'água expire, portanto, torne seu sink downstream idempotente para lidar com resultados atualizados. Use o modo de acréscimo para cargas de trabalho que o modo de atualização não suporta, como joins de transmissão-transmissão, ou quando você pode descartar dados que chegam atrasados. Não use o modo completo para baixa latência. Consulte Selecionar um modo de saída para Structured Streaming.
- Trigger : Use um trigger
processingTimecom um intervalo0, que inicia o próximo micro-batch assim que o anterior termina e novos dados estão disponíveis. Isso proporciona a menor latência de micro-lotes, mas aumenta os custos da API de armazenamento em cloud. Não useAvailableNow,OnceouContinuouspara cargas de trabalho operacionais. Consulte Configurar intervalos de trigger de Structured Streaming. - Marca d'água : defina a marca d'água com duração suficiente para incluir os dados que chegam atrasados e que sua carga de trabalho não deve descartar. A marca d'água controla por quanto tempo a query aceita dados de tempo de evento fora de ordem antes de descartá-los e remover o estado, portanto, uma marca d'água muito curta descarta silenciosamente registros válidos atrasados. Dentro dessa restrição, uma marca d'água mais curta reduz a latência e retém menos estado, e uma marca d'água mais longa tolera mais dados atrasados ao custo de latência e estado. Um pequeno múltiplo do seu SLA de latência, como 2x, é um ponto de partida razoável para o ajuste. Consulte Aplicar marcas d'água para controlar os limites de processamento de dados.
- Fontes e sinks : Leitura de fontes de baixa latência, como barramentos de mensagens (Apache Kafka, Amazon Kinesis, Apache Pulsar ou Google Cloud Pub/Sub) ou feeds de dados de alteração de tabelas Delta Lake e Apache Iceberg. Grave em coletores de baixa latência e alto throughput, como message buses, bancos de dados operacionais ou coletores
foreach. Projete operações de coletor para serem idempotentes, de modo que os consumidores downstream lidem com duplicidades e dados que chegam com atraso. - Estado e ponto de verificação : para consultas com estado, use o armazenamento de estado RocksDB, que é necessário tanto para o ponto de verificação de log de alterações quanto para o ponto de verificação de estado assíncrono. Habilite o ponto de verificação de log de alterações para persistir apenas alterações de estado incrementais. Quando o ponto de verificação de estado for o gargalo na duração do seu lote, habilite o ponto de verificação de estado assíncrono para sobrepor as gravações de ponto de verificação com o próximo micro-lote, após analisar suas ressalvas de recuperação de falhas e redimensionamento de cluster. Dê a cada query seu próprio diretório de ponto de verificação no armazenamento em cloud durável. Consulte Configurar o armazenamento de estado RocksDB no Databricks, Ponto de verificação de estado assíncrono para consultas com estado e Pontos de verificação do Structured Streaming.
- Gerenciamento de offset : Para reduzir a latência do checkpoint de offset em transmissões contínuas, habilite o acompanhamento de progresso assíncrono, que atualiza os logs de offset e commit sem bloquear o processamento de dados. Não é compatível com os triggers
AvailableNowouOnce. Consulte acompanhamento de progresso assíncrono. - Saltos de armazenamento : Mantenha a computação dentro de um único pipeline de transmissão sempre que possível. Dividir a lógica entre vários jobs ou pipelines adiciona saltos de armazenamento que aumentam a latência.
Projetar cargas de trabalho de transmissão para prever falhas
A Databricks recomenda que você sempre configure Jobs de transmissão para reiniciar automaticamente em caso de falha. Algumas capacidades, incluindo a evolução do esquema, exigem que as cargas de trabalho de transmissão estructurada tentem novamente automaticamente. Consulte Configurar jobs de Transmissão Estruturada para reiniciar consultas de transmissão em caso de falha.
Algumas operações como foreachBatch fornecem garantias de "pelo menos uma vez" em vez de "exatamente uma vez". Para essas operações, certifique-se de que seu pipeline de processamento seja idempotente. Consulte Usar foreachBatch para gravar em destinos de dados arbitrários.
Quando uma consulta é reiniciada, as microlotes planejadas durante os processos de execução anteriores. Se o trabalho falhou devido a um erro de falta de memória ou se o senhor cancelou manualmente um trabalho devido ao excesso de microlotes, talvez seja necessário escalonar o compute para processar com êxito os microlotes.
Se o senhor alterar as configurações entre as execuções, essas configurações serão aplicadas aos primeiros novos lotes planejados. Consulte Recuperação após alterações em uma consulta de transmissão estruturada.
Quando um job é repetido
O senhor pode programar várias tarefas como parte de um Databricks Job. Quando o senhor configura uma tarefa usando o acionador contínuo, não é possível definir dependências entre tarefas.
O senhor pode optar por programar várias transmissões em um único trabalho usando uma das seguintes abordagens:
- Tarefa múltipla : Definir um trabalho com várias tarefas que executam cargas de trabalho de transmissão usando o acionador contínuo.
- Várias consultas : Defina várias consultas de transmissão no código-fonte para uma única tarefa.
Você também pode combinar essas estratégias. A tabela a seguir compara essas abordagens.
Estratégia | Tarefa múltipla | Várias consultas |
|---|---|---|
Como o site compute é compartilhado? | Databricks recomenda implantar o Job compute adequadamente dimensionado para cada transmissão de tarefa. Opcionalmente, o senhor pode compartilhar o site compute entre as tarefas. | Todas as consultas compartilham o mesmo compute. Opcionalmente, você pode atribuir consultas ao pool de programadores. |
Como as novas tentativas são tratadas? | Todas as tarefas devem falhar antes que o trabalho tente novamente. | A tarefa tentará novamente se alguma consulta falhar. |
Para mais detalhes sobre como trabalhar com múltiplas tarefas ou consultas, veja Execução de Múltiplas Consultas de Transmissão Estruturada no Mesmo Cluster.
Configurar transmissão estruturada Job para reiniciar as consultas de transmissão em caso de falha
A Databricks recomenda que você configure todas as cargas de trabalho de transmissão usando o gatilho contínuo. Consulte Executar jobs continuamente.
O gatilho contínuo tem o seguinte comportamento por default:
- Evita mais de uma execução concorrente do trabalho.
- começar uma nova execução quando uma execução anterior falhar.
- Usa o recuo exponencial para novas tentativas.
Databricks recomenda sempre usar o Job compute em vez do compute para todos os fins ao programar fluxo de trabalho. Em caso de falha e nova tentativa de trabalho, novo compute recurso implantado.
A Databricks recomenda que você não use streamingQuery.awaitTermination() ou spark.streams.awaitAnyTermination(). Veja Quando usar awaitTermination().
Quando usar awaitTermination()
streamingQuery.awaitTermination() e spark.streams.awaitAnyTermination() bloqueiam a thread atual até que uma consulta de transmissão termine. A utilização dessas funções depende do seu ambiente de execução.
Para os Jobs do Lakeflow, não utilize streamingQuery.awaitTermination() ou spark.streams.awaitAnyTermination(). Estas funções não são necessárias porque o Serviço de Jobs impede automaticamente que uma execução seja concluída quando uma consulta de transmissão está ativa. Ambas as funções impedem que as células do Notebook sejam concluídas e impedem que o serviço Jobs faça o acompanhamento da consulta de transmissão, o que interrompe as métricas de backlog e as notificações de Job.
Utilize awaitTermination() nos seguintes casos:
Caso de uso | Comportamento |
|---|---|
Notebook interativo em computede uso geral |
|
Ambientes locais e de desenvolvimento | Ao executar um programa Spark localmente, o processo é encerrado quando a thread principal termina. Chame |
Propagação da falha para o driver | Sem |