Ejecución de consultas adaptables

La ejecución de consultas adaptables (AQE) es la reoptimización de consultas que se produce durante el procesamiento de consultas.

La motivación para la re-optimización en tiempo de ejecución es que Azure Databricks tiene las estadísticas más precisas actualizadas al final de un intercambio aleatorio y de difusión (denominado fase de consulta en AQE). Como resultado, Azure Databricks puede optar por una mejor estrategia física, elegir un tamaño y un número óptimos de partición posteriores a la mezcla, o realizar optimizaciones que solían requerir sugerencias, por ejemplo, la gestión de uniones sesgadas.

Esto puede ser muy útil cuando la recopilación de estadísticas no está activada o cuando las estadísticas están obsoletas. También es útil en lugares donde las estadísticas derivadas estáticamente son inexactas, como en medio de una consulta complicada, o después de la aparición de la asimetría de datos.

Capacidades

AQE está habilitado de forma predeterminada. Tiene cuatro características principales:

• Cambia dinámicamente la combinación de combinación de ordenación en combinación hash de difusión.
• Fusiona las particiones de manera dinámica (las particiones pequeñas en otras de tamaño razonable) después del intercambio aleatorio. Las tareas muy pequeñas tienen un rendimiento de E/S peor y tienden a sufrir más de la sobrecarga de programación y la sobrecarga de configuración de tareas. La combinación de tareas pequeñas ahorra recursos y mejora el rendimiento del clúster.
• Controla dinámicamente el sesgo en la fusión mediante combinación de ordenación y la combinación de hash aleatoria mediante la división (y replicación si es necesario) de las tareas sesgadas en tareas de tamaño uniforme aproximadamente.
• Detecta y propaga dinámicamente relaciones vacías.

Aplicación

AQE se aplica a todas las consultas que:

• No son de streaming
• Contienen al menos una operación de intercambio (normalmente cuando hay una unión, una agregación o una ventana), una subconsulta, o ambas cosas.

No todas las consultas aplicadas por AQE se vuelven a optimizar necesariamente. Es posible que la re-optimización genere un plan de consulta diferente al compilado de manera estática. Para determinar si AQE ha cambiado el plan de una consulta, consulte la sección siguiente, Planes de consulta.

Planes de consulta

En esta sección se describe cómo puede examinar los planes de consulta de diferentes maneras.

En esta sección:

• Interfaz de usuario de Spark
• DataFrame.explain()
• SQL EXPLAIN

Interfaz de usuario de Spark

AdaptiveSparkPlan nodo

Las consultas aplicadas a AQE contienen uno o varios AdaptiveSparkPlan nodos, normalmente como nodo raíz de cada consulta principal o subconsulta. Antes de que se ejecute la consulta o cuando se esté ejecutando, la isFinalPlan marca del nodo correspondiente AdaptiveSparkPlan se muestra como false; una vez completada la ejecución de la consulta, la isFinalPlan marca cambia a . true.

Plan en evolución

El diagrama del plan de consulta evoluciona a medida que avanza la ejecución y refleja el plan más actual que se está ejecutando. Los nodos que ya se han ejecutado (en los que las métricas están disponibles) no cambiarán, pero aquellos que no lo han hecho pueden cambiar con el tiempo al ser reoptimizados.

A continuación se muestra un ejemplo de diagrama de plan de consulta:

DataFrame.explain()

AdaptiveSparkPlan nodo

Las consultas aplicadas a AQE contienen uno o varios AdaptiveSparkPlan nodos, normalmente como nodo raíz de cada consulta principal o subconsulta. Antes de que se ejecute la consulta o cuando se ejecute, la isFinalPlan marca del nodo correspondiente AdaptiveSparkPlan se muestra como false; una vez completada la ejecución de la consulta, la isFinalPlan marca cambia a true.

Plan actual e inicial

En cada AdaptiveSparkPlan nodo habrá el plan inicial (el plan antes de aplicar las optimizaciones de AQE) y el plan actual o final, en función de si la ejecución se ha completado. El plan actual evolucionará a medida que avanza la ejecución.

Estadísticas de tiempo de ejecución

Cada fase de orden aleatorio y difusión contiene estadísticas de datos.

Antes de que se ejecute la etapa o durante su ejecución, las estadísticas son estimaciones en tiempo de compilación, y el indicador isRuntime es false, por ejemplo: Statistics(sizeInBytes=1024.0 KiB, rowCount=4, isRuntime=false);

Una vez completada la ejecución de la fase, las estadísticas son las recopiladas en tiempo de ejecución y la marca isRuntime se convertirá en true, por ejemplo: Statistics(sizeInBytes=658.1 KiB, rowCount=2.81E+4, isRuntime=true)

A continuación se muestra un DataFrame.explain ejemplo:

• Antes de la ejecución

  

• Durante la ejecución

  

• Después de la ejecución

  

SQL EXPLAIN

AdaptiveSparkPlan nodo

Las consultas aplicadas a AQE contienen uno o varios nodos AdaptiveSparkPlan, normalmente como nodo raíz de cada consulta principal o subconsulta.

Ningún plan actual

Como SQL EXPLAIN no ejecuta la consulta, el plan actual es siempre el mismo que el plan inicial y no refleja lo que finalmente se ejecutaría con AQE.

A continuación se muestra un ejemplo de explicación de SQL:

Efectividad

El plan de consulta cambiará si una o varias optimizaciones de AQE surten efecto. El efecto de estas optimizaciones de AQE se demuestra por la diferencia entre los planes actuales y finales y el plan inicial y los nodos de plan específicos en los planes actuales y finales.

• Cambio dinámico de la fusión mediante combinación de ordenación en combinación de hash de difusión: distintos nodos de combinación física entre el plan actual o final, y el plan inicial

  

• Fusión dinámica de particiones: nodo CustomShuffleReader con la propiedad Coalesced

  

  

• Control dinámico de la combinación de sesgo: nodo SortMergeJoin con el campo isSkew como true.

  

  

• Detectar y propagar dinámicamente relaciones vacías: parte (o la totalidad) del plan es reemplazada por el nodo LocalTableScan con el campo de relación vacío.

  

  

Configuración

En esta sección:

• Habilitación y deshabilitación de la ejecución de consultas adaptables
• Habilitar la reorganización optimizada automáticamente
• Cambiar dinámicamente una unión de ordenación y fusión a una unión hash de difusión
• Fusionar dinámicamente particiones
• Administrar dinámicamente las uniones sesgadas
• Detección y propagación dinámica de relaciones vacías

Habilitación y deshabilitación de la ejecución de consultas adaptables

Habilitación del orden aleatorio de optimización automática

Cambio dinámico de la fusión mediante combinación de ordenación a combinación de hash de difusión

Fusión dinámica de particiones

Control dinámico de la combinación de sesgo

Una partición se considera sesgada cuando ambos (partition size > skewedPartitionFactor * median partition size) y (partition size > skewedPartitionThresholdInBytes) son true.

Detección y propagación dinámica de relaciones vacías

Preguntas más frecuentes (FAQ)

En esta sección:

• ¿Por qué AQE no ha difundido una tabla de unión pequeña?
• ¿Debería seguir utilizando una sugerencia de estrategia de unión por difusión con AQE habilitado?
• ¿Cuál es la diferencia entre la sugerencia de combinación de sesgo y la optimización de combinación de sesgo de AQE? ¿Cuál debo usar?
• ¿Por qué AQE no ha ajustado automáticamente el orden de mi unión?
• ¿Por qué AQE no detectó mi asimetría de datos?

¿Por qué AQE no ha difundido una tabla de unión pequeña?

Si el tamaño de la relación que se espera que se difunda se encuentra por debajo de este umbral, pero todavía no se difunde:

• Compruebe el tipo de combinación. La difusión no se admite para determinados tipos de combinación, por ejemplo, la relación izquierda de una operación LEFT OUTER JOIN no se puede difundir.
• También es posible que la relación contenga una gran cantidad de particiones vacías, en cuyo caso la mayoría de las tareas pueden finalizar rápidamente con la fusión mediante combinación de ordenación o se puede optimizar con el control de la combinación sesgada. AQE evita cambiar estas fusiones mediante combinación de ordenación por combinaciones de hash de difusión si el porcentaje de particiones no vacías es inferior a spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin.

¿Se debe seguir usando una sugerencia de estrategia de combinación de difusión con AQE habilitado?

Sí. Una combinación de difusión planeada estáticamente suele ser más efectiva que una planeada de forma dinámica por parte de AQE, ya que es posible que AQE no cambie a la combinación de difusión hasta después de realizar el orden aleatorio para ambos lados de la combinación (momento en el que se obtienen los tamaños de relación reales). Por tanto, el uso de una sugerencia de difusión puede seguir siendo una buena opción si conoce bien la consulta. AQE respetará las sugerencias de consulta de la misma manera que la optimización estática, pero todavía puede aplicar optimizaciones dinámicas que no se ven afectadas por las sugerencias.

¿Cuál es la diferencia entre la sugerencia de combinación de sesgo y la optimización de combinación de sesgo de AQE? ¿Cuál debo usar?

Se recomienda confiar en el control de la combinación de sesgo de AQE en lugar de usar la sugerencia de combinación de sesgo, ya que la combinación de sesgo de AQE es completamente automática y, en general, funciona mejor que la de la sugerencia.

¿Por qué AQE no ajustó automáticamente mi orden de unión?

La reordenación de combinaciones dinámicas no forma parte de AQE.

¿Por qué AQE no detectó mi asimetría de datos?

Hay dos condiciones de tamaño que deben cumplirse para que AQE detecte una partición como una partición sesgada:

• El tamaño de la partición es mayor que ( spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes valor predeterminado de 256 MB)
• El tamaño de la partición es mayor que el tamaño medio de todas las particiones multiplicado por el factor de partición sesgada spark.sql.adaptive.skewJoin.skewedPartitionFactor (valor predeterminado: 5)

Además, el manejo del sesgo de AQE solo se aplica a las uniones basadas en barajado (uniones de ordenación y fusión y uniones de hash con barajado). Las uniones de difusión nunca se optimizan para el sesgo. Para las uniones soportadas, el tipo de unión determina qué lado puede optimizar AQE:

Por ejemplo, en un LEFT OUTER JOIN, solo se puede optimizar el sesgo del lado izquierdo, y un FULL OUTER JOIN no tiene optimización de sesgo en ninguno de sus lados.

Legado

El término "Ejecución adaptable" ha existido desde Spark 1.6, pero el nuevo AQE en Spark 3.0 es fundamentalmente diferente. En términos de funcionalidad, Spark 1.6 solo realiza la función de "agrupar dinámicamente particiones". En términos de arquitectura técnica, el nuevo AQE es un marco de planeamiento dinámico y replanado de consultas basadas en estadísticas en tiempo de ejecución, que admite una variedad de optimizaciones como las que hemos descrito en este artículo y se puede ampliar para habilitar más optimizaciones potenciales.