Compartir a través de

Control de consultas grandes en flujos de trabajo interactivos

Un desafío con los flujos de trabajo de datos interactivos es controlar consultas grandes. Esto incluye consultas que generan demasiadas filas de salida, capturan muchas particiones externas o calculan en conjuntos de datos extremadamente grandes. Estas consultas pueden ser extremadamente lentas, saturar los recursos de proceso y dificultar que otros compartan el mismo proceso.

Query Watchdog es un proceso que impide que las consultas monopolicen los recursos de cómputo, examinando las causas más comunes de consultas extensas y terminando las consultas que superan un umbral. En este artículo se describe cómo habilitar y configurar Query Watchdog.

Importante

Query Watchdog está habilitado para todas las computaciones de propósito general creadas mediante la interfaz de usuario.

Ejemplo de una consulta disruptiva

Un analista está realizando algunas consultas ad hoc en un almacenamiento de datos Just-In-Time. El analista usa un proceso de escalado automático compartido que facilita que varios usuarios usen un solo proceso al mismo tiempo. Supongamos que hay dos tablas que cada una tiene un millón de filas.

import org.apache.spark.sql.functions._
spark.conf.set("spark.sql.shuffle.partitions", 10)

spark.range(1000000)
  .withColumn("join_key", lit(" "))
  .createOrReplaceTempView("table_x")
spark.range(1000000)
  .withColumn("join_key", lit(" "))
  .createOrReplaceTempView("table_y")

Estos tamaños de tabla se pueden administrar en Apache Spark. Sin embargo, cada una incluye una join_key columna con una cadena vacía en cada fila. Esto puede ocurrir si los datos no están perfectamente limpios o si hay asimetría de datos significativa en la que algunas claves son más frecuentes que otras. Estas claves de combinación vacías son mucho más frecuentes que cualquier otro valor.

En el código siguiente, el analista está uniendo estas dos tablas en sus claves, lo que genera resultados de un billón de resultados, y todos ellos se generan en un único ejecutor (el ejecutor que obtiene la " " clave):

SELECT
  id, count(id)
FROM
  (SELECT
    x.id
  FROM
    table_x x
  JOIN
    table_y y
  on x.join_key = y.join_key)
GROUP BY id

Parece que esta consulta se está ejecutando. Sin embargo, sin saber sobre los datos, el analista ve que hay "solo" una única tarea que queda durante la ejecución del trabajo. La consulta nunca finaliza, dejando al analista frustrado y confuso sobre por qué no funcionó.

En este caso, solo hay una clave de combinación problemática. Otras veces puede haber muchas más.

Habilitación y configuración de Query Watchdog

Para habilitar y configurar Query Watchdog, se requieren los pasos siguientes.

  • Habilite Watchdog con spark.databricks.queryWatchdog.enabled.
  • Configure el entorno de ejecución de la tarea con spark.databricks.queryWatchdog.minTimeSecs.
  • Muestra la salida con spark.databricks.queryWatchdog.minOutputRows.
  • Configure la relación de salida con spark.databricks.queryWatchdog.outputRatioThreshold.

Para evitar que una consulta cree demasiadas filas de salida para el número de filas de entrada, puede habilitar Query Watchdog y configurar el número máximo de filas de salida como un múltiplo del número de filas de entrada. En este ejemplo se usa una proporción de 1000 (valor predeterminado).

spark.conf.set("spark.databricks.queryWatchdog.enabled", true)
spark.conf.set("spark.databricks.queryWatchdog.outputRatioThreshold", 1000L)

La última configuración declara que cualquier tarea determinada nunca debe generar más de 1000 veces el número de filas de entrada.

Sugerencia

La relación de salida es completamente personalizable. Se recomienda comenzar más abajo y ver qué umbral funciona bien para usted y su equipo. Un intervalo de 1000 a 10 000 es un buen punto de partida.

No solo Query Watchdog impide que los usuarios monopolicen los recursos de proceso para los trabajos que nunca se completarán, sino que también ahorra tiempo al fracasar rápidamente en una consulta que nunca se completaría. Por ejemplo, la consulta siguiente producirá un error después de varios minutos porque supera el límite.

SELECT
  z.id
  join_key,
  sum(z.id),
  count(z.id)
FROM
  (SELECT
    x.id,
    y.join_key
  FROM
    table_x x
  JOIN
    table_y y
  on x.join_key = y.join_key) z
GROUP BY join_key, z.id

Esto es lo que vería:

Monitor de consultas

Normalmente, es suficiente habilitar Query Watchdog y establecer la relación de umbral de salida/entrada, pero también tiene la opción de establecer dos propiedades adicionales: spark.databricks.queryWatchdog.minTimeSecs y spark.databricks.queryWatchdog.minOutputRows. Estas propiedades especifican el tiempo mínimo que debe ejecutar una tarea determinada en una consulta antes de cancelarla y el número mínimo de filas de salida de una tarea de esa consulta.

Por ejemplo, puede establecer minTimeSecs en un valor mayor si desea darle la oportunidad de generar un gran número de filas por tarea. Del mismo modo, debe establecer spark.databricks.queryWatchdog.minOutputRows a diez millones si desea detener una consulta solo después de que una tarea en esa consulta haya generado diez millones de filas. Cualquier valor inferior y la consulta tiene éxito, incluso si se superó la proporción de salida/entrada.

spark.conf.set("spark.databricks.queryWatchdog.minTimeSecs", 10L)
spark.conf.set("spark.databricks.queryWatchdog.minOutputRows", 100000L)

Sugerencia

Si configura Query Watchdog en un cuaderno, la configuración no se conserva en los reinicios de proceso. Si desea configurar Query Watchdog para todos los usuarios de un proceso, se recomienda usar una configuración de proceso.

Detección de consultas en conjuntos de datos extremadamente grandes

Otra consulta grande típica puede examinar una gran cantidad de datos de tablas o conjuntos de datos grandes. La operación de escaneo puede durar mucho tiempo y saturar los recursos de computación (incluso leer metadatos de una tabla de Hive grande puede tardar un período de tiempo significativo). Puede establecer maxHivePartitions para evitar obtener demasiadas particiones de una tabla grande de Hive. Del mismo modo, también puede establecer maxQueryTasks para limitar las consultas en un conjunto de datos extremadamente grande.

spark.conf.set("spark.databricks.queryWatchdog.maxHivePartitions", 20000)
spark.conf.set("spark.databricks.queryWatchdog.maxQueryTasks", 20000)

¿Cuándo debe habilitar Query Watchdog?

Query Watchdog debe estar habilitado para el proceso de análisis ad hoc en el que los analistas de SQL y los científicos de datos comparten un proceso determinado y un administrador debe asegurarse de que las consultas "juegan bien" entre sí.

¿Cuándo debe deshabilitar Query Watchdog?

En general, no recomendamos cancelar rápidamente las consultas utilizadas en un escenario de ETL, ya que normalmente no hay intervención humana para corregir el error. Se recomienda deshabilitar Query Watchdog excepto para el cálculo de analítica ad hoc.

  • Last updated on