Execução de consulta adaptável

A execução adaptativa de consultas (AQE) é a reotimização de consultas que ocorre durante a execução da consulta.

Antes de uma consulta ser executada, o otimizador escolhe um plano de consulta com base nas estimativas do tamanho dos dados. Essas estimativas podem estar erradas se as estatísticas das tabelas estiverem em falta ou obsoletas, se os dados estiverem enviesados ou em etapas posteriores de uma consulta complexa. Um plano baseado em estimativas erradas pode usar uma estratégia de junção lenta, criar demasiadas tarefas pequenas após um shuffle, ou criar algumas tarefas sobredimensionadas que atrasam toda a fase. O AQE resolve estes problemas ao re-otimizar o plano enquanto a consulta é executada, com base no tamanho real dos dados.

No final de cada troca de embaralhamento ou broadcast, o Azure Databricks apresenta estatísticas precisas para os dados produzidos até ao momento. O AQE utiliza estas estatísticas para escolher uma estratégia física melhor, escolher um número e tamanho ótimos de partições após o shuffle, ou aplicar otimizações que, de outra forma, requerem hints de consulta, como o tratamento de skew joins.

O AQE está habilitado por padrão. Não precisa de alterar as suas consultas para beneficiar disso. Verifique se a AQE alterou um plano de consulta ao diagnosticar uma consulta lenta. Ajusta as definições do AQE apenas quando os valores predefinidos não se ajustam à tua carga de trabalho. Consulte Planos de consulta e Configuração.

Capacidades

O AQE tem quatro características principais:

• Altera dinamicamente a junção de mesclagem de classificação na junção de hash de transmissão.
• Aglutina dinamicamente partições (combine pequenas partições em partições de tamanho razoável) após a troca aleatória. Tarefas muito pequenas têm pior taxa de transferência de E/S e tendem a sofrer mais com a sobrecarga de agendamento e a sobrecarga de configuração de tarefas. A combinação de pequenas tarefas economiza recursos e melhora a taxa de transferência do cluster.
• Manipula dinamicamente a inclinação na junção de mesclagem de classificação e a junção de hash aleatória dividindo (e replicando, se necessário) tarefas distorcidas em tarefas de tamanho aproximadamente uniforme.
• Deteta e propaga dinamicamente relações vazias, que são partes de uma consulta que não retornam linhas.

Consultas suportadas

O AQE aplica-se a consultas que cumpram ambas as condições:

• A consulta não é do tipo em fluxo.
• A consulta contém pelo menos um exchange (normalmente de uma junção, agregado ou janela), pelo menos uma subconsulta, ou ambas.

O AQE re-otimiza todas as consultas suportadas, mas o resultado pode ser o mesmo plano que o otimizador compilou antes da execução da consulta. Para verificar se a AQE alterou o plano de uma consulta, consulte Planos de consulta.

Planos de consulta

As consultas aplicadas ao AQE contêm um ou mais nós AdaptiveSparkPlan, geralmente servindo como o nó raiz de cada consulta principal ou subconsulta. Cada AdaptiveSparkPlan nodo tem um isFinalPlan indicador. A flag está false antes e durante a execução da consulta, e muda para true quando a consulta termina.

Para visualizar um plano de consulta, use a interface do Spark, DataFrame.explain(), ou SQL EXPLAIN. Cada um mostra informações diferentes sobre como a AQE altera o plano.

Interface do usuário do Spark

O diagrama do plano de consulta na interface do Spark mostra o plano mais atual e atualiza-se à medida que a consulta avança. Os nós que já foram executados e têm métricas não mudam. Os nós que ainda não foram executados podem mudar à medida que o AQE reotimiza o plano.

Este exemplo mostra um diagrama de plano de consulta:

DataFrame.explain()

DataFrame.explain() mostra dois planos sob cada nó AdaptiveSparkPlan: o plano inicial, antes de o AQE aplicar quaisquer otimizações, e o plano atual. O plano atual muda à medida que a consulta é executada e torna-se o plano final quando a consulta termina.

A saída inclui estatísticas de dados para cada shuffle e fase de broadcast. As estatísticas dependem de se a etapa já foi executada:

• Antes e durante a execução da etapa, isRuntime está false e as estatísticas são estimativas em tempo de compilação. Por exemplo, Statistics(sizeInBytes=1024.0 KiB, rowCount=4, isRuntime=false).
• Após a fase de execução, isRuntime está true e as estatísticas são medições de tempo de execução. Por exemplo, Statistics(sizeInBytes=658.1 KiB, rowCount=2.81E+4, isRuntime=true).

Estas capturas de ecrã mostram DataFrame.explain() a saída antes, durante e depois da execução:

• Antes da execução

  

• Durante a execução

  

• Após a execução

  

SQL EXPLAIN

SQL EXPLAIN não executa a consulta, por isso o AQE não pode re-otimizá-la. O resultado mostra apenas o plano inicial, que pode diferir do plano que a AQE executa.

Este exemplo mostra SQL EXPLAIN a saída:

Identificar otimizações aplicadas num plano de consulta

Quando o AQE aplica uma otimização, o plano de consulta muda. Para identificar quais as otimizações que tiveram efeito, compare o plano atual ou final com o plano inicial. Cada otimização altera o plano de uma forma específica:

• Alteração da estratégia de associação O plano atual ou final tem um nó físico de junção diferente do plano inicial.

  

• Consolidação de partições. O plano contém um CustomShuffleReader nó com a Coalesced propriedade.

  

  

• Processamento de skew join. O plano contém um nó SortMergeJoin com o campo isSkew definido como true.

  

  

• Propagação de relação vazia. Um LocalTableScan nó com um campo de relação vazio substitui parte ou a totalidade do plano.

  

  

Configuração

Estas propriedades de configuração do Apache Spark controlam o AQE. A maioria das cargas de trabalho não precisam de alterar as predefinições.

Ativar o AQE e ajustar as otimizações individuais

Utilize estas propriedades para ligar ou desligar o AQE e para ajustar o shuffle auto-otimizado, a alternância de broadcast join e a propagação de relações vazias:

Propriedade	Tipo	Default	Description
spark.databricks.optimizer.adaptive.enabled	Boolean	true	Ativa ou desativa a execução adaptativa de consultas.
spark.sql.shuffle.partitions	Integer	200	O número padrão de partições a serem usadas ao embaralhar dados para junções ou agregações. Defina-o para auto para permitir o embaralhamento auto-otimizado, que define este número com base no plano de consulta e no tamanho dos dados de entrada da consulta. Para Structured Streaming, não podes alterar este valor entre reinícios de consulta a partir do mesmo ponto de controlo.
spark.databricks.adaptive.autoBroadcastJoinThreshold	Byte String	30MB	O limiar de tamanho para mudar uma junção de intercalação por ordenação para uma junção de hash broadcast em tempo de execução.
spark.databricks.adaptive.emptyRelationPropagation.enabled	Boolean	true	Ativa ou desativa a propagação dinâmica de relações vazias.

Aglutinar partições dinamicamente

Propriedade	Tipo	Default	Description
spark.sql.adaptive.coalescePartitions.enabled	Boolean	true	Liga ou desliga a coalescência de partições.
spark.sql.adaptive.advisoryPartitionSizeInBytes	Byte String	64MB	O tamanho do alvo após a coalescência. As partições coalescidas são próximas, mas não maiores do que este tamanho.
spark.sql.adaptive.coalescePartitions.minPartitionSize	Byte String	1MB	O tamanho mínimo das partições após a coalescência. As partições coalescidas não são menores do que este tamanho.
spark.sql.adaptive.coalescePartitions.minPartitionNum	Integer	Duas vezes o número de núcleos de cluster	O número mínimo de partições após a coalescência. Não definas esta propriedade, porque defini-la sobrepõe o valor de spark.sql.adaptive.coalescePartitions.minPartitionSize.

Manipule dinamicamente a junção inclinada

Propriedade	Tipo	Default	Description
spark.sql.adaptive.skewJoin.enabled	Boolean	true	Ativa ou desativa o manuseamento de skew join.
spark.sql.adaptive.skewJoin.skewedPartitionFactor	Integer	5	O multiplicador para o tamanho mediano da partição que uma partição deve exceder para ser considerada enviesada.
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes	Byte String	256MB	O tamanho mínimo que uma partição deve ultrapassar para ser considerada enviesada.

AQE trata uma partição como desbalanceada apenas quando ambas as condições são verdadeiras: (partition size > skewedPartitionFactor * median partition size) e (partition size > skewedPartitionThresholdInBytes).

Perguntas frequentes (FAQ)

Porque a AQE não transmitiu uma pequena tabela de junção?

Se a tabela for menor do que spark.databricks.adaptive.autoBroadcastJoinThreshold mas a AQE não a difunde, verifique estas causas:

• O tipo de junção não suporta broadcast. Alguns tipos de junção não conseguem difundir um dos lados. Por exemplo, o AQE não pode transmitir o lado esquerdo de um LEFT OUTER JOIN.
• A tabela tem muitas divisões vazias. A maioria das tarefas na junção sort-merge termina rapidamente, ou o tratamento da junção com distribuição enviesada otimiza a junção. O AQE não altera uma junção por fusão ordenada para uma junção de hash broadcast se a percentagem de partições não vazias for inferior a spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin.

Devo continuar a usar uma dica de broadcast join com o AQE ativado?

Sim, se conheces bem a tua consulta. Uma junção broadcast que o otimizador planeia antes de a query correr geralmente tem um desempenho melhor do que uma que o AQE planeia em tempo de execução. O AQE pode não mudar para uma junção por difusão até depois de baralhar ambos os lados da junção, altura em que aprende os tamanhos reais das tabelas.

O AQE respeita os hints de consulta da mesma forma que a otimização estática faz, e ainda aplica as otimizações dinâmicas que os hints não afetam.

Devo usar o hint de skew join ou o tratamento de skew join do AQE?

Usa o manuseamento de skew join do AQE. É automático e normalmente tem um desempenho melhor do que a dica de skew join.

Porque é que a AQE não reordenou os meus joins?

O AQE não reordena as junções em tempo de execução. Para melhorar a ordem de junção, recolha estatísticas da tabela para que o otimizador baseado em custos possa escolher um plano melhor. Veja Otimizador baseado em custos.

Por que a AQE não detetou minha distorção de dados?

O AQE trata uma partição como enviesada apenas quando ambas as condições são verdadeiras:

• A partição é maior que spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (por defeito 256MB).
• A partição é maior do que spark.sql.adaptive.skewJoin.skewedPartitionFactor (por defeito 5) vezes o tamanho mediano da partição.

O AQE também lida com o skew apenas em joins baseados em shuffle, que são joins sort-merge e shuffle hash joins. Nunca otimiza o desequilíbrio nas junções de difusão. Para junções baseadas em shuffle, o tipo de junção determina que lado o AQE pode otimizar:

Tipo de junção	Inclinação otimizada no lado esquerdo	Inclinação otimizada para o lado direito
INNER	Yes	Yes
CROSS	Yes	Yes
LEFT OUTER	Yes	Não
RIGHT OUTER	Não	Yes
LEFT SEMI	Yes	Não
LEFT ANTI	Yes	Não
FULL OUTER	Não	Não

Legado

O Spark 1.6 introduziu uma funcionalidade chamada execução adaptativa, mas o AQE no Spark 3.0 é uma funcionalidade diferente. A execução adaptativa do Spark 1.6 apenas agrega as partições dinamicamente. O AQE é um framework que planeia e replaneia consultas com base em estatísticas de tempo de execução, e suporta todas as otimizações mencionadas nesta página.