Acelerar Google Cloud Serverless para Apache Spark con Lightning Engine

En este documento se explica cómo habilitar Lightning Engine para acelerar las cargas de trabajo por lotes y las sesiones interactivas de Serverless para Apache Spark.

Información general

Lightning Engine es un acelerador de consultas de alto rendimiento que se basa en un motor de optimización multicapa que lleva a cabo técnicas de optimización habituales, como las optimizaciones de consultas y de ejecución, así como optimizaciones seleccionadas en la capa del sistema de archivos y en los conectores de acceso a datos.

Como se muestra en la siguiente ilustración, Lightning Engine acelera el rendimiento de la ejecución de consultas de Spark en una carga de trabajo similar a TPC-H (tamaño del conjunto de datos 10 TB).

Para obtener más información, consulta el artículo Presentamos Lightning Engine, la nueva generación del rendimiento de Apache Spark.

Disponibilidad de Lightning Engine

  • Lightning Engine se puede usar con los entornos de ejecución de Apache Spark sin servidores admitidos que están disponibles de forma general (actualmente, los entornos de ejecución 1.2, 2.2 y 2.3; no está disponible en el entorno de ejecución de Spark 3.0).
  • Lightning Engine solo está disponible con el nivel de precios premium de Serverless para Apache Spark.
    • Cargas de trabajo por lotes: Lightning Engine se habilita automáticamente para las cargas de trabajo por lotes en el nivel Premium. No tienes que hacer nada.
    • Sesiones interactivas: Lightning Engine no está habilitado de forma predeterminada para las sesiones interactivas. Para habilitarlo, consulta el artículo Habilitar Lightning Engine.
    • Plantillas de sesión: Lightning Engine no está habilitado de forma predeterminada en las plantillas de sesión. Para habilitarlo, consulta el artículo Habilitar Lightning Engine.

Habilitar Lightning Engine

En las siguientes secciones se muestra cómo habilitar Lightning Engine en una carga de trabajo por lotes, una plantilla de sesión y una sesión interactiva de Serverless para Apache Spark.

Carga de trabajo por lotes

Habilitar Lightning Engine en una carga de trabajo por lotes

Puedes usar la Google Cloud consola, la CLI de Google Cloud o la API de Dataproc para habilitar Lightning Engine en una carga de trabajo por lotes.

Consola

Usa la Google Cloud consola para habilitar Lightning Engine en una carga de trabajo por lotes.

  1. En la Google Cloud consola:

    1. Ve a Lotes de Dataproc.
    2. Haz clic en Crear para abrir la página Crear lote.

  2. Selecciona y rellena los siguientes campos:

    • Contenedor:

    • Configuración de niveles:

      • Selecciona Premium. De esta forma, se habilita y se marca automáticamente la opción "Habilitar el motor Lightning para acelerar el rendimiento de Spark".

      Cuando seleccionas el nivel Premium, los campos Nivel de computación del controlador y Nivel de computación del ejecutor se definen como Premium. Estos ajustes de computación de nivel premium, que se definen automáticamente, no se pueden anular en lotes que usen tiempos de ejecución anteriores a 3.0.

      Puede configurar Driver Disk Tier y Executor Disk Tier en Premium o dejar el valor predeterminado Standard. Si eliges un nivel de disco premium, debes seleccionar el tamaño del disco. Para obtener más información, consulta las propiedades de asignación de recursos.

    • Propiedades: opcional: introduce el par Key (nombre de la propiedad) y Value si quieres seleccionar el tiempo de ejecución Native Query Execution:

      Clave Valor
      spark.dataproc.lightningEngine.runtime nativo

  3. Rellena, selecciona o confirma otros ajustes de cargas de trabajo en lote. Consulta Enviar una carga de trabajo por lotes de Spark.

  4. Haz clic en Enviar para ejecutar la carga de trabajo por lotes de Spark.

gcloud

Define las siguientes marcas de comando de gcloud CLI gcloud dataproc batches submit spark para habilitar Lightning Engine en una carga de trabajo por lotes.

gcloud dataproc batches submit spark \
    --project=PROJECT_ID \
    --region=REGION \
    --properties=dataproc.tier=premium \
    OTHER_FLAGS_AS_NEEDED

Notas:

  • PROJECT_ID: tu ID de proyecto Google Cloud . Los IDs de proyecto se indican en la sección Información del proyecto del panel de control de la consola Google Cloud .
  • REGION: una región de Compute Engine disponible para ejecutar la carga de trabajo.

  • --properties=dataproc.tier=premium. Al definir el nivel premium, se definen automáticamente las siguientes propiedades en la carga de trabajo por lotes:

    • spark.dataproc.engine=lightningEngine selecciona Lightning Engine para la carga de trabajo por lotes.
    • spark.dataproc.driver.compute.tier y spark.dataproc.executor.compute.tier se definen como premium (consulta las propiedades de asignación de recursos). Estos ajustes de computación de nivel premium, que se definen automáticamente, no se pueden anular en las solicitudes por lotes que usen tiempos de ejecución anteriores a 3.0.

  • Otras propiedades

    • Native Query Engine: spark.dataproc.lightningEngine.runtime=native añade esta propiedad si quieres seleccionar el tiempo de ejecución de Native Query Execution.

    • Niveles y tamaños de disco: de forma predeterminada, los tamaños de disco del controlador y del ejecutor se definen en standard niveles y tamaños. Puede añadir propiedades para seleccionar premium niveles y tamaños de disco (en múltiplos de 375 GiB).
      Para obtener más información, consulte las propiedades de asignación de recursos.

  • OTHER_FLAGS_AS_NEEDED: consulta Enviar una carga de trabajo por lotes de Spark.

API

Para habilitar Lightning Engine en una carga de trabajo por lotes, añade "dataproc.tier":"premium" a RuntimeConfig.properties como parte de tu solicitud batches.create. Al definir el nivel premium, se definen automáticamente las siguientes propiedades en la carga de trabajo por lotes:

  • spark.dataproc.engine=lightningEngine selecciona Lightning Engine para la carga de trabajo por lotes.
  • spark.dataproc.driver.compute.tier y spark.dataproc.executor.compute.tier se definen como premium (consulta las propiedades de asignación de recursos). Estos ajustes de computación de nivel premium, que se definen automáticamente, no se pueden anular en las tandas que usen tiempos de ejecución anteriores a 3.0.

Otro RuntimeConfig.properties:

  • Native Query Engine: spark.dataproc.lightningEngine.runtime:native. Añada esta propiedad si quiere seleccionar el tiempo de ejecución de Native Query Execution.

  • Niveles y tamaños de disco: de forma predeterminada, los tamaños de disco del controlador y del ejecutor se definen en standard niveles y tamaños. Puede añadir propiedades para seleccionar premium niveles y tamaños (en múltiplos de 375 GiB).
    Para obtener más información, consulte las propiedades de asignación de recursos.

Consulta Enviar una carga de trabajo por lotes de Spark para definir otros campos de la API de cargas de trabajo por lotes.

Plantilla de sesión

Habilitar Lightning Engine en una plantilla de sesión

Puedes usar la Google Cloud consola, la CLI de Google Cloud o la API de Dataproc para habilitar Lightning Engine en una plantilla de sesión de Jupyter o Spark Connect.

Consola

Usa la Google Cloud consola para habilitar Lightning Engine en una carga de trabajo por lotes.

  1. En la Google Cloud consola:

    1. Ve a Plantillas de sesión de Dataproc.
    2. Haga clic en Crear para abrir la página Crear plantilla de sesión.

  2. Selecciona y rellena los siguientes campos:

    • Información de la plantilla de sesión:
      • Selecciona "Habilitar Lightning Engine para acelerar el rendimiento de Spark".
    • Configuración de ejecución:

    • Propiedades: Introduce los siguientes pares Key (nombre de la propiedad) y Value para seleccionar el nivel Premium:

      Clave Valor
      dataproc.tier premium
      spark.dataproc.engine lightningEngine

      Opcional: Introduce el siguiente par Key (nombre de la propiedad) y Value para seleccionar el tiempo de ejecución de Native Query Execution:

      Clave Valor
      spark.dataproc.lightningEngine.runtime native

  3. Rellena, selecciona o confirma otros ajustes de la plantilla de sesión. Consulta Crear una plantilla de sesión.

  4. Haga clic en Enviar para crear la plantilla de sesión.

gcloud

No puedes crear directamente una plantilla de sesión de Serverless para Apache Spark con la CLI de gcloud. En su lugar, puedes usar el comando gcloud beta dataproc session-templates import para importar una plantilla de sesión, editarla para habilitar Lightning Engine y, opcionalmente, el tiempo de ejecución de consultas nativas, y, a continuación, exportar la plantilla editada con el comando gcloud beta dataproc session-templates export.

API

Para habilitar Lightning Engine en una plantilla de sesión, añade "dataproc.tier":"premium" y "spark.dataproc.engine":"lightningEngine" a RuntimeConfig.properties como parte de tu solicitud sessionTemplates.create.

Otro RuntimeConfig.properties:

  • Native Query Engine: spark.dataproc.lightningEngine.runtime:native: Añade esta propiedad a RuntimeConfig.properties para seleccionar el tiempo de ejecución de Native Query Execution.

Consulta Crear una plantilla de sesión para definir otros campos de la API de plantillas de sesión.

Sesión interactiva

Habilitar Lightning Engine en una sesión interactiva

Puedes usar la CLI de Google Cloud o la API de Dataproc para habilitar Lightning Engine en una sesión interactiva de Serverless para Apache Spark. También puedes habilitar Lightning Engine en una sesión interactiva en un cuaderno de BigQuery Studio.

gcloud

Define las siguientes marcas de comando de la CLI de gcloud gcloud beta dataproc sessions create spark para habilitar Lightning Engine en una sesión interactiva.

gcloud beta dataproc sessions create spark \
    --project=PROJECT_ID \
    --location=REGION \
    --properties=dataproc.tier=premium,spark.dataproc.engine=lightningEngine \
    OTHER_FLAGS_AS_NEEDED

Notas:

  • PROJECT_ID: tu ID de proyecto Google Cloud . Los IDs de proyecto se indican en la sección Información del proyecto del panel de control de la consola Google Cloud .
  • REGION: una región de Compute Engine disponible para ejecutar la carga de trabajo.

  • --properties=dataproc.tier=premium,spark.dataproc.engine=lightningEngine. Estas propiedades habilitan Lightning Engine en la sesión.

  • Otras propiedades:

    • Native Query Engine: spark.dataproc.lightningEngine.runtime=native: añade esta propiedad para seleccionar el tiempo de ejecución de Native Query Execution.

  • OTHER_FLAGS_AS_NEEDED: consulta Crear una sesión interactiva.

API

Para habilitar Lightning Engine en una sesión, añade "dataproc.tier":"premium" y "spark.dataproc.engine":"lightningEngine" a RuntimeConfig.properties como parte de tu solicitud sessions.create.

Otro RuntimeConfig.properties:

* Native Query Engine: spark.dataproc.lightningEngine.runtime:native: Añade esta propiedad a RuntimeConfig.properties si quieres seleccionar el tiempo de ejecución Native Query Execution.

Consulta Crear una sesión interactiva para definir otros campos de la API de plantilla de sesión.

Notebook de BigQuery

Puedes habilitar Lightning Engine al crear una sesión en un cuaderno de PySpark de BigQuery Studio. 

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session
session = Session()

# Enable Lightning Engine.
session.runtime_config.properties["dataproc.tier"] = "premium"
session.runtime_config.properties["spark.dataproc.engine"] = "lightningEngine"

# Enable THE Native Query Execution runtime.
session.runtime_config.properties["spark.dataproc.lightningEngine.runtime"] = "native"

# Create the Spark session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate())

# Add Spark application code here:

Verificar la configuración de Lightning Engine

Puedes usar la Google Cloud consola, la CLI de Google Cloud o la API de Dataproc para verificar la configuración de Lightning Engine en una carga de trabajo por lotes, una plantilla de sesión o una sesión interactiva.

Carga de trabajo por lotes

  • Para verificar que el nivel del lote se ha definido como premium y que el motor se ha definido como Lightning Engine, sigue estos pasos:

    • Google Cloud Consola: en la página Lotes, consulta las columnas Nivel y Motor del lote. También puede hacer clic en ID de lote para ver estos ajustes en la página de detalles del lote.
    • CLI de gcloud: ejecuta el comando gcloud dataproc batches describe.
    • API: envía una solicitud batches.get.

Plantilla de sesión

  • Para verificar que el motor esté configurado como Lightning Engine en una plantilla de sesión, sigue estos pasos:

    • Google Cloud Consola: en la página Plantillas de sesión, consulta la columna Engine de tu plantilla. También puede hacer clic en el Nombre de la plantilla de sesión para ver este ajuste en la página de detalles de la plantilla de sesión.
    • CLI de gcloud: ejecuta el comando gcloud beta dataproc session-templates describe.
    • API: envía una solicitud sessionTemplates.get.

Sesión interactiva

  • El motor se establece en Lightning Engine para una sesión interactiva:

    • Google Cloud Consola: en la página Sesiones interactivas, consulta la columna Engine (Motor) de la plantilla. Puedes hacer clic en el ID de sesión interactiva para ver este ajuste en la página de detalles de la plantilla de sesión.
    • CLI de gcloud: ejecuta el comando gcloud beta dataproc sessions describe.
    • API: envía una solicitud sessions.get.

Ejecución de consultas nativas

La ejecución de consultas nativas (NQE) es una función opcional de Lightning Engine que mejora el rendimiento mediante una implementación nativa basada en Apache Gluten y Velox, diseñada para el hardware de Google.

El tiempo de ejecución de consultas nativas incluye una gestión de memoria unificada para cambiar dinámicamente entre la memoria fuera del montículo y la memoria en el montículo sin tener que modificar las configuraciones de Spark. NQE incluye una mayor compatibilidad con operadores, funciones y tipos de datos de Spark, así como inteligencia para identificar automáticamente oportunidades de usar el motor nativo para optimizar las operaciones de inserción.

Identificar cargas de trabajo de ejecución de consultas nativas

Utilice la ejecución de consultas nativas en los siguientes casos:

  • APIs de DataFrame de Spark, APIs de Dataset de Spark y consultas de Spark SQL que leen datos de archivos Parquet y ORC. El formato del archivo de salida no afecta al rendimiento de la ejecución de consultas nativas.

  • Cargas de trabajo recomendadas por la herramienta de cualificación de ejecución de consultas nativas.

No se recomienda ejecutar consultas nativas en cargas de trabajo con entradas de los siguientes tipos de datos:

  • Byte: ORC y Parquet
  • Marca de tiempo: ORC
  • Estructura, matriz y mapa: Parquet

Limitaciones de la ejecución de consultas nativas

Si habilitas la ejecución de consultas nativas en los siguientes casos, pueden producirse excepciones, incompatibilidades con Spark o que la carga de trabajo vuelva al motor de Spark predeterminado.

Alternativas

La ejecución de consultas nativas en la siguiente ejecución puede provocar que la carga de trabajo vuelva al motor de ejecución de Spark, lo que puede provocar regresiones o errores.

  • ANSI: si el modo ANSI está habilitado, la ejecución vuelve a Spark.

  • Modo que distingue entre mayúsculas y minúsculas: la ejecución de consultas nativas solo admite el modo predeterminado de Spark, que no distingue entre mayúsculas y minúsculas. Si el modo de distinción entre mayúsculas y minúsculas está habilitado, se pueden producir resultados incorrectos.

  • Análisis de tablas con particiones: la ejecución de consultas nativas solo admite el análisis de tablas con particiones cuando la ruta contiene la información de la partición. De lo contrario, la carga de trabajo vuelve al motor de ejecución de Spark.

Comportamiento incompatible

Si se usa la ejecución de consultas nativas en los siguientes casos, se pueden producir comportamientos incompatibles o resultados incorrectos:

  • Funciones JSON: la ejecución de consultas nativas admite cadenas entre comillas dobles, no entre comillas simples. Se producen resultados incorrectos con comillas simples. Si se usa "*" en la ruta con la función get_json_object, se devuelve NULL.

  • Configuración de lectura de Parquet:

    • La ejecución de consultas nativas trata spark.files.ignoreCorruptFiles como si tuviera el valor predeterminado false, aunque se haya definido como true.
    • La ejecución de consultas nativas ignora spark.sql.parquet.datetimeRebaseModeInRead y devuelve solo el contenido del archivo Parquet. No se tienen en cuenta las diferencias entre el calendario híbrido antiguo (juliano gregoriano) y el calendario gregoriano proléptico. Los resultados de Spark pueden variar.

  • NaN: no se admite. Por ejemplo, se pueden producir resultados inesperados al usar NaN en una comparación numérica.

  • Lectura columnar de Spark: se puede producir un error grave porque el vector columnar de Spark no es compatible con la ejecución de consultas nativas.

  • Desbordamiento: si las particiones aleatorias se definen con un número elevado, la función de desbordamiento en disco puede activar un OutOfMemoryException. Si esto ocurre, reducir el número de particiones puede eliminar esta excepción.