Execução de consultas adaptável

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

A motivação para a reotimização do tempo de execução é que o Azure Databricks tem as estatísticas precisas mais atualizadas no final de uma troca aleatória e de transmissão (referida como um estágio de consulta no AQE). Como resultado, o Azure Databricks pode optar por uma estratégia física melhor, escolher um tamanho e número de partição pós-embaralhamento ideais ou fazer otimizações que costumavam exigir dicas, por exemplo, manipulação de junção inclinada.

Isto pode ser muito útil quando a recolha de estatísticas não está ativada ou quando as estatísticas estão obsoletas. Também é útil em locais onde as estatísticas derivadas estaticamente são imprecisas, como no meio de uma consulta complicada ou após a ocorrência de distorção de dados.

Capacidades

O AQE está habilitado por padrão. Tem 4 características principais:

• Altera dinamicamente a associação de intercalação de ordenação para associação hash de difusã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.
• Processa dinamicamente a distorção na associação de intercalação de ordenação e associação hash aleatória ao dividir (e substituir, se necessário) tarefas distorcidas em tarefas com um tamanho aproximadamente uniforme.
• Deteta e propaga dinamicamente relações vazias.

Aplicação

A AQE aplica-se a todas as consultas que sejam:

• Não transmissão em fluxo
• Contêm pelo menos uma troca (geralmente quando há uma junção, agregação ou janela), uma subconsulta ou ambas.

Nem todas as consultas aplicadas ao AQE são necessariamente reotimizadas. A reotimização pode ou não criar um plano de consulta diferente daquele compilado estaticamente. Para determinar se o plano de uma consulta foi alterado pela AQE, consulte a seção a seguir, Planos de consulta.

Planos de consulta

Esta seção discute como você pode examinar planos de consulta de diferentes maneiras.

Nesta secção:

• Interface do usuário do Spark
• DataFrame.explain()
• SQL EXPLAIN

IU do Apache Spark

Nó AdaptiveSparkPlan

As consultas aplicadas ao AQE contêm um ou mais AdaptiveSparkPlan nós, geralmente como o nó raiz de cada consulta ou subconsulta principal. Antes de a consulta ser executada ou quando estiver em execução, o isFinalPlan sinalizador do nó correspondente AdaptiveSparkPlan aparece como false; após a conclusão da execução da consulta, o isFinalPlan sinalizador muda para true.

Plano evolutivo

O diagrama de plano de consulta evolui à medida que a execução progride e reflete o plano mais atual que está sendo executado. Os nós que já foram executados (nos quais as métricas estão disponíveis) não serão alterados, mas aqueles que não foram executados podem mudar ao longo do tempo como resultado de reotimizações.

Segue-se um exemplo de diagrama de plano de consulta:

DataFrame.explain()

Nó AdaptiveSparkPlan

As consultas aplicadas ao AQE contêm um ou mais AdaptiveSparkPlan nós, geralmente como o nó raiz de cada consulta ou subconsulta principal. Antes da execução da consulta ou quando ela está em execução, o isFinalPlan sinalizador do nó correspondente AdaptiveSparkPlan é exibido como false; após a conclusão da execução da consulta, o isFinalPlan sinalizador muda para true.

Plano atual e plano inicial

Em cada AdaptiveSparkPlan nó haverá o plano inicial (o plano antes de aplicar quaisquer otimizações de AKE) e o plano atual ou final, dependendo se a execução foi concluída. O plano atual evoluirá à medida que a execução progride.

Estatísticas de tempo de execução

Cada etapa de shuffle e broadcast contém estatísticas de dados.

Antes da execução do estágio ou quando o estágio está em execução, as estatísticas são estimativas de tempo de compilação, e o sinalizador isRuntime é false, por exemplo: Statistics(sizeInBytes=1024.0 KiB, rowCount=4, isRuntime=false);

Após a conclusão da execução do estágio, as estatísticas são aquelas coletadas em tempo de execução, e o sinalizador isRuntime se tornará true, por exemplo: Statistics(sizeInBytes=658.1 KiB, rowCount=2.81E+4, isRuntime=true)

Segue-se um DataFrame.explain exemplo:

• Antes da execução

  

• Durante a execução

  

• Após a execução

  

SQL EXPLAIN

Nó AdaptiveSparkPlan

As consultas aplicadas ao AQE contêm um ou mais nós AdaptiveSparkPlan, geralmente como o nó raiz de cada consulta ou subconsulta principal.

Nenhum plano atual

Como SQL EXPLAIN não executa a consulta, o plano atual é sempre o mesmo que o plano inicial e não reflete o que acabaria por ser executado pela AQE.

Segue-se um exemplo de explicação SQL:

Eficácia

O plano de consulta será alterado se uma ou mais otimizações AQE entrarem em vigor. O efeito dessas otimizações de AQE é demonstrado pela diferença entre os planos atual e final e os nós do plano inicial e do plano específico nos planos atual e final.

• Alterar dinamicamente a associação de intercalação de ordenação para associação hash de difusão: nós de associação físicos diferentes entre o plano atual/final e o plano inicial

  

• Aglutinar partições dinamicamente: nó CustomShuffleReader com propriedade Coalesced

  

  

• Manipule dinamicamente a junção de inclinação: nó SortMergeJoin com campo isSkew como verdadeiro.

  

  

• Detetar e propagar dinamicamente relações vazias: parte (ou inteira) do plano é substituída pelo nó LocalTableScan com o campo de relação como vazio.

  

  

Configuração

Nesta secção:

• Habilitar e desabilitar a execução de consultas adaptáveis
• Habilite o shuffle otimizado automaticamente
• Alterar dinamicamente a associação de intercalação de ordenação para associação hash de difusão
• Aglutinar partições dinamicamente
• Manipule dinamicamente a junção inclinada
• Detetar e propagar dinamicamente relações vazias

Habilitar e desabilitar a execução de consultas adaptáveis

Habilite o shuffle otimizado automaticamente

Alterar dinamicamente a associação de intercalação de ordenação para associação hash de difusão

Aglutinar partições dinamicamente

Manipule dinamicamente a junção inclinada

Uma partição é considerada enviesada quando ambos (partition size > skewedPartitionFactor * median partition size) e (partition size > skewedPartitionThresholdInBytes) são true.

Detetar e propagar dinamicamente relações vazias

Perguntas mais frequentes (FAQ)

Nesta secção:

• Por que a AQE não transmitiu uma pequena mesa de junção?
• Ainda devo usar uma dica de estratégia de ingresso de transmissão com o AQE ativado?
• Qual é a diferença entre a dica de junção inclinada e a otimização de junção de distorção AKE? Qual devo usar?
• Porque é que a AQE não ajustou a minha encomenda de adesão automaticamente?
• Por que a AQE não detetou minha distorção de dados?

Por que a AQE não transmitiu uma pequena mesa de junção?

Se a dimensão da relação que se espera que seja difundida for inferior a este limiar, mas continuar a não ser difundida:

• Verifique o tipo de junção. A transmissão não é suportada para certos tipos de junção, por exemplo, a relação esquerda de um LEFT OUTER JOIN não pode ser transmitida.
• Também pode ser que a relação contenha muitas partições vazias, caso em que a maioria das tarefas pode terminar rapidamente com junção de mesclagem de classificação ou pode ser potencialmente otimizada com manipulação de junção inclinada. O AQE evitará a alteração destas associações de intercalação de ordenação para associações hash de difusão se a percentagem de partições não vazias for inferior a spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin.

Ainda devo usar uma dica de estratégia de ingresso de transmissão com o AQE ativado?

Sim. Uma junção de transmissão planejada estaticamente é geralmente mais performante do que uma unida planejada dinamicamente pela AKE, pois a AQE pode não mudar para a junção de transmissão até depois de executar o shuffle para ambos os lados da junção (quando os tamanhos de relação reais são obtidos). Portanto, usar uma dica de transmissão ainda pode ser uma boa escolha se você conhecer bem sua consulta. O AQE respeitará as dicas de consulta da mesma forma que a otimização estática, mas ainda poderá aplicar otimizações dinâmicas que não são afetadas pelas dicas.

Qual é a diferença entre a dica de junção inclinada e a otimização de junção de distorção AKE? Qual devo usar?

Recomenda-se confiar no manuseio de junção de inclinação AQE em vez de usar a dica de junção inclinada, porque a junção de inclinação AQE é completamente automática e, em geral, tem um desempenho melhor do que a contraparte de dica.

Porque é que a AQE não ajustou a minha encomenda de adesão automaticamente?

A reordenação dinâmica da junção não faz parte da AQE.

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

Há duas condições de tamanho que devem ser satisfeitas para que o AQE detete uma partição como uma partição enviesada:

• O tamanho da partição é maior do que o spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (padrão 256MB)
• O tamanho da partição é maior do que o tamanho médio de todas as partições vezes o fator spark.sql.adaptive.skewJoin.skewedPartitionFactor de partição distorcida (padrão 5)

Além disso, o suporte de manipulação de distorção é limitado para certos tipos de junção, por exemplo, no LEFT OUTER JOIN, apenas a inclinação no lado esquerdo pode ser otimizada.

Legado

O termo "Execução Adaptativa" existe desde o Spark 1.6, mas o novo AQE no Spark 3.0 é fundamentalmente diferente. Em termos de funcionalidade, o Spark 1.6 faz apenas a parte de "coalescência dinâmica de partições". Em termos de arquitetura técnica, o novo AQE é uma estrutura de planejamento dinâmico e replanejamento de consultas com base em estatísticas de tempo de execução, que suporta uma variedade de otimizações como as que descrevemos neste artigo e pode ser estendida para permitir mais otimizações potenciais.