Managed Service para Apache Spark

1. Descripción general: Serverless for Apache Spark

Managed Service para Apache Spark es un servicio completamente administrado y con alto escalamiento para ejecutar Apache Spark, Apache Flink, Presto y muchas otras herramientas y frameworks de código abierto. Usa Apache Spark administrado para la modernización de data lakes, ETL / ELT y ciencia de datos segura a escala mundial. Managed Apache Spark también está completamente integrado en varios servicios de Google Cloud, incluidos BigQuery, Cloud Storage, Gemini Enterprise Agent Engine y Knowledge Catalog.

Apache Spark administrado está disponible en dos versiones:

• Managed Apache Spark sin servidores te permite ejecutar trabajos de PySpark sin necesidad de configurar la infraestructura ni el ajuste de escala automático. Managed Apache Spark admite cargas de trabajo por lotes y sesiones o notebooks de PySpark.
• Los clústeres administrados de Apache Spark te permiten administrar un clúster de Hadoop YARN para cargas de trabajo de Spark basadas en YARN, además de herramientas de código abierto, como Flink y Presto. Puedes personalizar tus clústeres basados en la nube con el escalamiento vertical u horizontal que desees, incluido el ajuste de escala automático.

En este codelab, aprenderás varias formas diferentes de consumir Dataproc Serverless.

Apache Spark se creó originalmente para ejecutarse en clústeres de Hadoop y se usó YARN como su administrador de recursos. Mantener los clústeres de Hadoop requiere un conjunto específico de conocimientos y garantizar que muchos parámetros diferentes de los clústeres estén configurados correctamente. Esto se suma a un conjunto independiente de parámetros de configuración que Spark también requiere que el usuario establezca. Esto genera muchas situaciones en las que los desarrolladores pasan más tiempo configurando su infraestructura que trabajando en el código de Spark en sí.

Dataproc Serverless elimina la necesidad de configurar manualmente los clústeres de Hadoop o Spark. Dataproc Serverless no se ejecuta en Hadoop y usa su propia asignación dinámica de recursos para determinar sus requisitos de recursos, incluido el ajuste de escala automático. Un pequeño subconjunto de propiedades de Spark aún se puede personalizar con Dataproc Serverless, pero, en la mayoría de los casos, no será necesario modificarlas.

2. Configurar

Comenzarás por configurar tu entorno y los recursos que se usan en este codelab.

Crea un proyecto de Google Cloud. Puedes usar uno existente.

Haz clic en Cloud Shell en la barra de herramientas de Cloud Console para abrirlo.

Cloud Shell proporciona un entorno de Shell listo para usar que puedes utilizar en este codelab.

Cloud Shell establecerá el nombre del proyecto de forma predeterminada. Ejecuta echo $GOOGLE_CLOUD_PROJECT para verificar. Si no ves el ID de tu proyecto en el resultado, configúralo.

export GOOGLE_CLOUD_PROJECT=<your-project-id>

Configura una región de Compute Engine para tus recursos, como us-central1 o europe-west2.

export REGION=<your-region>

Habilita las APIs

En este codelab, se usan las siguientes APIs:

• BigQuery
• Dataproc

Habilita las APIs necesarias. Este proceso tardará aproximadamente un minuto y, cuando se complete, aparecerá un mensaje de éxito.

gcloud services enable bigquery.googleapis.com
gcloud services enable dataproc.googleapis.com

Configurar el acceso a la red

Dataproc Serverless requiere que habilites el Acceso privado a Google en la región en la que ejecutarás tus trabajos de Spark, ya que los controladores y ejecutores de Spark solo tienen IPs privadas. Ejecuta el siguiente comando para habilitarlo en la subred default.

gcloud compute networks subnets update default \
  --region=${REGION} \
  --enable-private-ip-google-access

Puedes verificar que Google Private Access esté habilitado con el siguiente comando, que mostrará True o False.

gcloud compute networks subnets describe default \
  --region=${REGION} \
  --format="get(privateIpGoogleAccess)"

Crea un bucket de almacenamiento

Crea un bucket de almacenamiento que se usará para almacenar los recursos creados en este codelab.

Elige un nombre para tu bucket. Los nombres de los buckets deben ser únicos a nivel global para todos los usuarios.

export BUCKET=<your-bucket-name>

Crea el bucket en la región en la que planeas ejecutar tus trabajos de Spark.

gsutil mb -l ${REGION} gs://${BUCKET}

Puedes ver que tu bucket está disponible en la consola de Cloud Storage. También puedes ejecutar gsutil ls para ver tu bucket.

Crea un servidor de historial persistente

La IU de Spark proporciona un amplio conjunto de herramientas de depuración y estadísticas sobre los trabajos de Spark. Para ver la IU de Spark de los trabajos completados de Dataproc Serverless, debes crear un clúster de Dataproc de un solo nodo para utilizarlo como servidor de historial persistente.

Establece un nombre para tu servidor de historial persistente.

PHS_CLUSTER_NAME=my-phs

Ejecuta el siguiente comando.

gcloud dataproc clusters create ${PHS_CLUSTER_NAME} \
    --region=${REGION} \
    --single-node \
    --enable-component-gateway \
    --properties=spark:spark.history.fs.logDirectory=gs://${BUCKET}/phs/*/spark-job-history

La IU de Spark y el servidor de historial persistente se explorarán con más detalle más adelante en el codelab.

3. Ejecuta trabajos de Spark sin servidores con Dataproc Batches

En este ejemplo, trabajarás con un conjunto de datos del conjunto de datos públicos Citi Bike Trips de la ciudad de Nueva York (NYC). NYC Citi Bikes es un sistema pagado de bicicletas compartidas de la ciudad de Nueva York. Realizarás algunas transformaciones simples y, luego, imprimirás los diez IDs de estaciones de Citi Bike más populares. Este ejemplo también usa notablemente el spark-bigquery-connector de código abierto para leer y escribir datos sin problemas entre Spark y BigQuery.

Clona el siguiente repositorio de GitHub y usa cd para ir al directorio que contiene el archivo citibike.py.

git clone https://github.com/GoogleCloudPlatform/devrel-demos.git
cd devrel-demos/data-analytics/next-2022-workshop/dataproc-serverless

citibike.py

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import BooleanType

if len(sys.argv) == 1:
    print("Please provide a GCS bucket name.")

bucket = sys.argv[1]
table = "bigquery-public-data:new_york_citibike.citibike_trips"

spark = SparkSession.builder \
          .appName("pyspark-example") \
          .config("spark.jars","gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar") \
          .getOrCreate()

df = spark.read.format("bigquery").load(table)

top_ten = df.filter(col("start_station_id") \
            .isNotNull()) \
            .groupBy("start_station_id") \
            .count() \
            .orderBy("count", ascending=False) \
            .limit(10) \
            .cache()

top_ten.show()

top_ten.write.option("header", True).csv(f"gs://{bucket}/citibikes_top_ten_start_station_ids")

Envía el trabajo a Serverless Spark con el SDK de Cloud, disponible en Cloud Shell de forma predeterminada. Ejecuta el siguiente comando en tu shell, que utiliza el SDK de Cloud y la API de Dataproc Batches para enviar trabajos de Spark sin servidores.

gcloud dataproc batches submit pyspark citibike.py \
  --batch=citibike-job \
  --region=${REGION} \
  --deps-bucket=gs://${BUCKET} \
  --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar \
--history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER_NAME} \
  -- ${BUCKET}

Para explicarlo mejor, veamos los siguientes puntos:

• gcloud dataproc batches submit hace referencia a la API de Dataproc Batches.
• pyspark indica que envías un trabajo de PySpark.
• --batch es el nombre del trabajo. Si no se proporciona, se usará un UUID generado de forma aleatoria.
• --region=${REGION} es la región geográfica en la que se procesará el trabajo.
• --deps-bucket=${BUCKET} es el lugar donde se sube tu archivo local de Python antes de ejecutarse en el entorno de Serverless.
• --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar incluye el archivo jar para el spark-bigquery-connector en el entorno de ejecución de Spark.
• --history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER} es el nombre completamente calificado del servidor de historial persistente. Aquí se almacenan los datos de eventos de Spark (separados del resultado de la consola) y se pueden ver desde la IU de Spark.
• El -- final indica que todo lo que se encuentre después de este serán argumentos de tiempo de ejecución para el programa. En este caso, enviarás el nombre de tu bucket, como lo requiere el trabajo.

Cuando se envíe el lote, verás el siguiente resultado.

Batch [citibike-job] submitted.

Después de unos minutos, verás el siguiente resultado junto con los metadatos del trabajo.

+----------------+------+
|start_station_id| count|
+----------------+------+
|             519|551078|
|             497|423334|
|             435|403795|
|             426|384116|
|             293|372255|
|             402|367194|
|             285|344546|
|             490|330378|
|             151|318700|
|             477|311403|
+----------------+------+

Batch [citibike-job] finished.

En la siguiente sección, aprenderás a ubicar los registros de este trabajo.

Funciones adicionales

Con Spark Serverless, tienes opciones adicionales para ejecutar tus trabajos.

• Puedes crear una imagen de Docker personalizada en la que se ejecute tu trabajo. Esta es una excelente manera de incluir dependencias adicionales, incluidas las bibliotecas de Python y R.
• Puedes conectar una instancia de Dataproc Metastore a tu trabajo para acceder a los metadatos de Hive.
• Para obtener más control, Dataproc Serverless admite la configuración de un pequeño conjunto de propiedades de Spark.

4. Métricas y observabilidad de Dataproc

En la consola de lotes de Dataproc, se enumeran todos tus trabajos de Dataproc Serverless. En la consola, verás el ID del lote, la ubicación, el estado, la hora de creación, el tiempo transcurrido y el tipo de cada trabajo. Haz clic en el ID del lote de tu trabajo para ver más información sobre él.

En esta página, verás información como Supervisión, que muestra cuántos ejecutores de Spark por lotes usó tu trabajo con el paso del tiempo (lo que indica cuánto se ajustó la escala automáticamente).

En la pestaña Detalles, verás más metadatos sobre el trabajo, incluidos los argumentos y parámetros que se enviaron con él.

También puedes acceder a todos los registros desde esta página. Cuando se ejecutan trabajos de Dataproc Serverless, se generan tres conjuntos diferentes de registros:

• A nivel del servicio
• Resultado de la consola
• Registro de eventos de Spark

A nivel del servicio: Incluye los registros que generó el servicio de Dataproc Serverless, como las solicitudes de Dataproc Serverless para obtener CPUs adicionales para el ajuste de escala automático. Para verlos, haz clic en Ver registros, que abrirá Cloud Logging.

Console output se puede ver en Output. Este es el resultado que genera el trabajo, incluidos los metadatos que Spark imprime cuando comienza un trabajo o cualquier instrucción de impresión incorporada en el trabajo.

Se puede acceder al registro de eventos de Spark desde la IU de Spark. Como proporcionaste a tu trabajo de Spark un servidor de historial persistente, puedes acceder a la IU de Spark haciendo clic en Ver servidor de historial de Spark, que contiene información sobre los trabajos de Spark que ejecutaste anteriormente. Puedes obtener más información sobre la IU de Spark en la documentación oficial de Spark.

5. Plantillas de Dataproc: BQ -> GCS

Las plantillas de Dataproc son herramientas de código abierto que ayudan a simplificar aún más las tareas de procesamiento de datos en la nube. Estos sirven como wrapper para Dataproc sin servidores y, además, incluyen plantillas para muchas tareas de importación y exportación de datos, incluidas las siguientes:

• BigQuerytoGCS y GCStoBigQuery
• GCStoBigTable
• GCStoJDBC y JDBCtoGCS
• HivetoBigQuery
• MongotoGCS y GCStoMongo

La lista completa está disponible en el archivo README.

En esta sección, usarás las plantillas de Dataproc para exportar datos de BigQuery a GCS.

Clone el repositorio

Clona el repositorio y cambia a la carpeta python.

git clone https://github.com/GoogleCloudPlatform/dataproc-templates.git
cd dataproc-templates/python

Configura el entorno

Ahora establecerás variables de entorno. Las plantillas de Dataproc usan la variable de entorno GCP_PROJECT para el ID de tu proyecto, así que configúrala como GOOGLE_CLOUD_PROJECT..

export GCP_PROJECT=${GOOGLE_CLOUD_PROJECT}

Tu región debe establecerse en el entorno desde antes. De lo contrario, configúralo aquí.

export REGION=<region>

Las plantillas de Dataproc usan el spark-bigquery-connector para procesar los trabajos de BigQuery y requieren que el URI se incluya en una variable de entorno JARS. Configura la variable JARS.

export JARS="gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar"

Configura los parámetros de la plantilla

Establece el nombre de un bucket de etapa de pruebas para que lo use el servicio.

export GCS_STAGING_LOCATION=gs://${BUCKET}

A continuación, establecerás algunas variables específicas del trabajo. Para la tabla de entrada, volverás a hacer referencia al conjunto de datos de BigQuery NYC Citibike.

BIGQUERY_GCS_INPUT_TABLE=bigquery-public-data.new_york_citibike.citibike_trips

Puedes elegir csv, parquet, avro o json. Para este codelab, elige CSV. En la siguiente sección, se explica cómo usar las plantillas de Dataproc para convertir tipos de archivos.

BIGQUERY_GCS_OUTPUT_FORMAT=csv

Establece el modo de salida en overwrite. Puedes elegir entre overwrite, append, ignore o errorifexists..

BIGQUERY_GCS_OUTPUT_MODE=overwrite

Configura la ubicación de salida de GCS como una ruta de acceso en tu bucket.

BIGQUERY_GCS_OUTPUT_LOCATION=gs://${BUCKET}/BQtoGCS

Ejecuta la plantilla

Ejecuta la plantilla BIGQUERYTOGCS especificándola a continuación y proporcionando los parámetros de entrada que configuraste.

./bin/start.sh \
-- --template=BIGQUERYTOGCS \
        --bigquery.gcs.input.table=${BIGQUERY_GCS_INPUT_TABLE} \
        --bigquery.gcs.output.format=${BIGQUERY_GCS_OUTPUT_FORMAT} \
        --bigquery.gcs.output.mode=${BIGQUERY_GCS_OUTPUT_MODE} \
        --bigquery.gcs.output.location=${BIGQUERY_GCS_OUTPUT_LOCATION}

El resultado será bastante ruidoso, pero después de aproximadamente un minuto verás lo siguiente.

Batch [5766411d6c78444cb5e80f305308d8f8] submitted.
...
Batch [5766411d6c78444cb5e80f305308d8f8] finished.

Para verificar que se generaron los archivos, ejecuta el siguiente comando.

gsutil ls ${BIGQUERY_GCS_OUTPUT_LOCATION}

De forma predeterminada, Spark escribe en varios archivos, según la cantidad de datos. En este caso, verás aproximadamente 30 archivos generados. Los nombres de los archivos de salida de Spark tienen el formato part, seguido de un número de cinco dígitos (que indica el número de parte) y una cadena de hash. Para grandes cantidades de datos, Spark suele escribir en varios archivos. Un ejemplo de nombre de archivo es part-00000-cbf69737-867d-41cc-8a33-6521a725f7a0-c000.csv.

6. Plantillas de Dataproc: De CSV a Parquet

Ahora usarás las plantillas de Dataproc para convertir datos en GCS de un tipo de archivo a otro con GCSTOGCS. Esta plantilla usa SparkSQL y proporciona la opción de enviar también una consulta de SparkSQL para que se procese durante la transformación para un procesamiento adicional.

Confirma las variables de entorno

Confirma que GCP_PROJECT, REGION y GCS_STAGING_BUCKET estén configuradas desde la sección anterior.

echo ${GCP_PROJECT}
echo ${REGION}
echo ${GCS_STAGING_LOCATION}

Establecer parámetros de plantilla

Ahora, establecerás los parámetros de configuración para GCStoGCS. Comienza con la ubicación de los archivos de entrada. Ten en cuenta que se trata de un directorio y no de un archivo específico, ya que se procesarán todos los archivos del directorio. Establece este parámetro en BIGQUERY_GCS_OUTPUT_LOCATION.

GCS_TO_GCS_INPUT_LOCATION=${BIGQUERY_GCS_OUTPUT_LOCATION}

Establece el formato del archivo de entrada.

GCS_TO_GCS_INPUT_FORMAT=csv

Establece el formato de salida deseado. Puedes elegir parquet, json, avro o csv.

GCS_TO_GCS_OUTPUT_FORMAT=parquet

Establece el modo de salida en overwrite. Puedes elegir entre overwrite, append, ignore o errorifexists..

GCS_TO_GCS_OUTPUT_MODE=overwrite

Establece la ubicación de salida.

GCS_TO_GCS_OUTPUT_LOCATION=gs://${BUCKET}/GCStoGCS

Ejecuta la plantilla

Ejecuta la plantilla GCStoGCS.

./bin/start.sh \
-- --template=GCSTOGCS \
        --gcs.to.gcs.input.location=${GCS_TO_GCS_INPUT_LOCATION} \
        --gcs.to.gcs.input.format=${GCS_TO_GCS_INPUT_FORMAT} \
        --gcs.to.gcs.output.format=${GCS_TO_GCS_OUTPUT_FORMAT} \
        --gcs.to.gcs.output.mode=${GCS_TO_GCS_OUTPUT_MODE} \
        --gcs.to.gcs.output.location=${GCS_TO_GCS_OUTPUT_LOCATION}

El resultado será bastante ruidoso, pero, después de aproximadamente un minuto, deberías ver un mensaje de confirmación como el siguiente.

Batch [c198787ba8e94abc87e2a0778c05ec8a] submitted.
...
Batch [c198787ba8e94abc87e2a0778c05ec8a] finished.

Para verificar que se generaron los archivos, ejecuta el siguiente comando.

gsutil ls ${GCS_TO_GCS_OUTPUT_LOCATION}

Con esta plantilla, también tienes la opción de proporcionar consultas de SparkSQL pasando gcs.to.gcs.temp.view.name y gcs.to.gcs.sql.query a la plantilla, lo que permite que se ejecute una consulta de SparkSQL en los datos antes de escribirlos en GCS.

7. Limpia los recursos

Para evitar que se apliquen cargos innecesarios a tu cuenta de GCP después de completar este codelab, haz lo siguiente:

1. Borra el bucket de Cloud Storage para el entorno que creaste.

gsutil rm -r gs://${BUCKET}

2. Borra el clúster de Dataproc que se usó para tu servidor de historial persistente.

gcloud dataproc clusters delete ${PHS_CLUSTER_NAME} \
  --region=${REGION}

3. Borra los trabajos de Dataproc Serverless. Ve a la consola de lotes, haz clic en la casilla junto a cada trabajo que quieras borrar y, luego, haz clic en BORRAR.

Si creaste un proyecto solo para este codelab, también puedes borrarlo de forma opcional:

1. En GCP Console, ve a la página Proyectos.
2. En la lista de proyectos, selecciona el proyecto que deseas borrar y haz clic en Borrar.
3. En el cuadro, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrarlo.

Precaución: Borrar un proyecto tiene las siguientes consecuencias:

• Se borra todo en el proyecto. Si utilizaste un proyecto existente para este instructivo, cuando lo borres, también borrarás cualquier otro trabajo que hayas realizado en el proyecto.
• Se pierden los ID personalizados de proyectos. Cuando creaste este proyecto, es posible que hayas creado un ID del proyecto personalizado que desees utilizar en el futuro. Para conservar las URL que usan el ID del proyecto, como una URL appspot.com, borra los recursos seleccionados dentro del proyecto en lugar de borrar todo el proyecto.

8. ¿Qué sigue?

En los siguientes recursos, se proporcionan formas adicionales de aprovechar Serverless Spark:

• Aprende a organizar flujos de trabajo de Dataproc Serverless con Cloud Composer.
• Aprende a integrar Dataproc Serverless con canalizaciones de Kubeflow.