Executar várias consultas de Streaming Estruturado no mesmo cluster

Muitos clientes executam várias consultas de Streaming Estruturado no mesmo cluster de Azure Databricks. Embora esse padrão tenha suporte, o Databricks recomenda limitar o número de consultas por cluster para evitar problemas de dimensionamento e gargalos de desempenho. Na computação sem servidor, Azure Databricks gerencia o dimensionamento automaticamente, portanto, essas considerações são tratadas para você. Se você estiver usando a computação clássica, em que controla o dimensionamento do driver e do executor, esta página descreve os principais gargalos para ter em mente e maneiras de resolvê-los.

Quando usar várias consultas no mesmo cluster

A execução de várias consultas de streaming no mesmo cluster reduz os custos de infraestrutura, especialmente quando você tem muitos fluxos pequenos que não exigem computação dedicada. O principal trade-off é o risco compartilhado de falha: se o cluster falhar, todos os streams nesse cluster falharão. Para pipelines de missão crítica, esse modo de falha comum geralmente é inaceitável.

Para cargas de trabalho que misturam fluxos críticos e não críticos, o Databricks recomenda o seguinte:

• Atribua a cada fluxo uma prioridade com base em seu impacto nos negócios.
• Coloque fluxos críticos em clusters dedicados, mesmo a um custo mais alto.
• Co-loque fluxos de menor prioridade para compartilhar recursos computacionais e reduzir custos.

Dimensionamento do driver

O driver é um recurso compartilhado. Várias consultas compartilham a mesma CPU, memória, agendador de DAG, agendador de tarefas e execução de UDF do lado do driver (por exemplo, foreachBatch). Ao executar muitos fluxos simultâneos, observe esses gargalos específicos além do provisionamento padrão de CPU e memória:

• Sobrecarga do Carregador Automático: se os fluxos usarem o Carregador Automático, a descoberta de arquivos e a listagem de diretórios aumentarão a pressão do driver.
• Limites de recursos no nível do sistema operacional (arquivos abertos): executar um alto volume de fluxos baseados em arquivo (como FileStreamSource ou Carregador Automático) simultaneamente em um único driver pode esgotar os limites do descritor de arquivo aberto no nível do usuário, o que pode causar falhas aleatórias no fluxo.
• Contrapressão no barramento de ouvinte: um grande número de consultas de streaming simultâneas pode causar contrapressão no barramento StreamingQueryListener na sessão única do Spark. Todos os eventos (incluindo onQueryIdle) são enviados para esse único barramento, e uma grande lista de pendências de eventos pode atrasar significativamente os manipuladores onQueryProgress assíncronos e afetar a estabilidade dos clusters.
• Operações dispendiosas no driver: evite chamar collect() ou outras operações dispendiosas do DataFrame no driver, a menos que seja absolutamente necessário, para evitar a materialização de grandes conjuntos de resultados e a geração de erros de memória insuficiente (OOM).

Solucionar problemas de contenção de drivers

Se você estiver enfrentando falhas em drivers devido a problemas de OOM ou contenção:

1. Monitore as métricas do driver na IU do Spark. Se você vir alto uso de CPU, memória ou disco, ajuste o dimensionamento do driver nas configurações de computação do cluster.
2. Se os problemas persistirem, verifique se o código não está executando operações com uso intensivo de memória ou UDFs no driver.
3. Se você não puder mais escalar o driver de forma vertical, o Databricks recomenda fortemente dividir seus trabalhos entre vários clusters para contornar esses gargalos de escalonamento de nós compartilhados.

Dimensionamento do executor

Com várias consultas em execução no mesmo cluster, todas as consultas compartilham slots de tarefa nos executores. Estágios de uma consulta podem ocupar slots disponíveis, levando a atrasos e privação para outras consultas. O Spark usa um mapeamento 1:1 entre slots de tarefa e núcleos disponíveis. Verifique se há núcleos suficientes disponíveis se as consultas precisarem ser executadas simultaneamente.

Em geral, os executores podem realizar operações com maior uso intensivo de memória do que o nó do driver. Ajuste a JVM do executor e os parâmetros de alocação de memória off-heap, se necessário, para lidar com a carga do aplicativo. Verifique se os nós executores estão dimensionados adequadamente em termos de CPU, memória e espaço em disco e faça o escalonamento vertical, se necessário. Se o dimensionamento vertical não for possível, considere adicionar nós de trabalho adicionais ao cluster.

Usar conjuntos de agendamento

Você pode configurar pools de agendadores para atribuir capacidade de computação a consultas ao executar várias consultas de streaming do mesmo código-fonte.

Por padrão, todas as consultas iniciadas em um notebook são executadas no mesmo pool de agendamento justo. Os trabalhos do Apache Spark gerados por gatilhos de todas as consultas de streaming em um notebook são executados um após o outro na ordem "primeiro a entrar, primeiro a sair" (FIFO). Isso pode causar atrasos desnecessários nas consultas, pois elas não estão compartilhando com eficiência os recursos do cluster.

Os pools de agendador permitem declarar quais consultas de Streaming Estruturado compartilham recursos de computação.

O exemplo a seguir atribui query1 a um pool dedicado, enquanto query2 e query3 compartilha um pool de agendador.

:::note compatibilidade Sem servidor

A Databricks recomenda deixar de usar spark.sparkContext, pois ele não é compatível com a arquitetura de computação sem servidor da Databricks. Use spark (SparkSession) diretamente em vez disso. Os conjuntos de recursos do agendador são um conceito de computação tradicional; no modo sem servidor, a Databricks gerencia o escalonamento e a alocação de recursos automaticamente.

:::

# Run streaming query1 in scheduler pool1
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool1")
df.writeStream.queryName("query1").toTable("table1")

# Run streaming query2 in scheduler pool2
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool2")
df.writeStream.queryName("query2").toTable("table2")

# Run streaming query3 in scheduler pool2
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool2")
df.writeStream.queryName("query3").toTable("table3")

Para obter mais informações sobre pools de agendadores justos, consulte a documentação de agendadores justos do Apache Spark.

Considerações de consulta com estado

Para consultas com estado em execução no mesmo cluster, tenha em mente o seguinte:

• Use o RocksDB como provedor de armazenamento de estado para evitar problemas de OOM e pausas de GC. O RocksDB é o provedor de armazenamento de estado padrão no Databricks Runtime 17.3 e superior. Confira Configurar o repositório de estado do RocksDB no Azure Databricks.
• Ajuste as partições de shuffle de acordo com os requisitos do seu aplicativo. Para estágios com estado, o Spark programa tarefas proporcionais ao número de partições aleatórias.
• Limite o uso de memória do RocksDB por nó para evitar erros de OOM causados pelo uso de memória off-heap. Isso é tratado automaticamente no Databricks Runtime 17.3 e superior, mas requer configuração manual em versões anteriores. Consulte o uso de memória do Cap RocksDB.
• Evite empacotar partições demais no mesmo nó executor. As operações de manutenção no repositório de estados, incluindo upload e limpeza de instantâneos, são executadas individualmente em cada nó. Atribuir partições demais a um nó de execução pode causar escassez de manutenção e tempos de recuperação mais longos devido à menor disponibilidade de snapshots completos.