Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
En esta página, se describe cómo usar Cloud Composer 2 para ejecutar cargas de trabajo Dataproc sin servidores en Google Cloud.
En los ejemplos de las siguientes secciones, se muestra cómo usar operadores para administrar cargas de trabajo por lotes de Dataproc Serverless. Usa estos operadores en DAG que crean, borran, enumeran y obtienen una carga de trabajo por lotes de Spark sin servidores de Dataproc:
Crea DAG para operadores que trabajan con cargas de trabajo de lotes sin servidores de Dataproc:
Crea DAG que usen contenedores personalizados y Dataproc Metastore.
Configura el Servidor de historial persistente para estos DAG.
Antes de comenzar
Habilita la API de Dataproc:
Console
Habilita la API de Dataproc.
gcloud
Habilita la API de Dataproc:
gcloud services enable dataproc.googleapis.com
Selecciona la ubicación del archivo de carga de trabajo por lotes. Puedes usar cualquiera de las siguientes opciones:
- Crea un bucket de Cloud Storage que almacene este archivo.
- Usa el bucket de tu entorno. Como no necesitas sincronizar este archivo con Airflow, puedes crear una subcarpeta separada fuera de las carpetas
/dags
o/data
. Por ejemplo,/batches
. - Usa un bucket existente.
Configura archivos y variables de Airflow
En esta sección, se muestra cómo configurar archivos y variables de Airflow para este instructivo.
Sube un archivo de carga de trabajo de Spark ML sin servidores de Dataproc a un bucket
La carga de trabajo de este instructivo ejecuta una secuencia de comandos pyspark:
Guarda cualquier secuencia de comandos pyspark en un archivo local llamado
spark-job.py
. Por ejemplo, puedes usar la secuencia de comandos de pyspark de muestra.Sube el archivo a la ubicación que seleccionaste en Antes de comenzar.
Configura variables de Airflow
En los ejemplos de las siguientes secciones, se usan variables de Airflow. Establece valores para estas variables en Airflow y, luego, tu código DAG podrá acceder a estos valores.
En los ejemplos de este instructivo, se usan las siguientes variables de Airflow. Puedes configurarlas según sea necesario, según el ejemplo que uses.
Configura las siguientes variables de Airflow para usarlas en tu código de DAG:
project_id
: ID del proyecto.bucket_name
: El URI de un bucket en el que se encuentra el archivo principal de Python de la carga de trabajo (spark-job.py
). Seleccionaste esta ubicación en Antes de comenzar.phs_cluster
: Es el nombre del clúster del servidor de historial persistente. Debes establecer esta variable cuando creas un servidor de historial persistente.image_name
: Es el nombre y la etiqueta de la imagen de contenedor personalizada (image:tag
). Establece esta variable cuando usas una imagen de contenedor personalizada con DataprocCreateBatchOperator.metastore_cluster
: Es el nombre del servicio de Dataproc Metastore. Establece esta variable cuando usas el servicio de Dataproc Metastore con DataprocCreateBatchOperator.region_name
: Es la región en la que se encuentra el servicio de Dataproc Metastore. Establece esta variable cuando usas el servicio de Dataproc Metastore con DataprocCreateBatchOperator.
Usa la consola de Google Cloud y la IU de Airflow para configurar cada variable de Airflow
En la consola de Google Cloud, ve a la página Entornos.
En la lista de entornos, haz clic en el vínculo Airflow de tu entorno. Se abrirá la IU de Airflow.
En la IU de Airflow, selecciona Administrador > Variables.
Haz clic en Add a new record.
Especifica el nombre de la variable en el campo Key y establece su valor en el campo Val.
Haz clic en Guardar.
Crea un servidor de historial persistente
Usa un servidor de historial persistente (PHS) para ver los archivos de historial de Spark de tus cargas de trabajo por lotes:
- Crea un servidor de historial persistente.
- Asegúrate de haber especificado el nombre del clúster de PHS en la variable de Airflow
phs_cluster
.
DataprocCreateBatchOperator
El siguiente DAG inicia una carga de trabajo de Dataproc Serverless Batch.
Para obtener más información sobre los argumentos DataprocCreateBatchOperator
, consulta el código fuente del operador.
Para obtener más información sobre los atributos que puedes pasar en el parámetro batch
de DataprocCreateBatchOperator
, consulta la descripción de la clase Batch.
Usa una imagen de contenedor personalizada con DataprocCreateBatchOperator
En el siguiente ejemplo, se muestra cómo usar una imagen de contenedor personalizada para ejecutar tus cargas de trabajo. Puedes usar un contenedor personalizado, por ejemplo, para agregar dependencias de Python no proporcionadas por la imagen de contenedor predeterminada.
Para usar una imagen de contenedor personalizada, sigue estos pasos:
Crea una imagen de contenedor personalizada y súbela a Container Registry.
Especifica la imagen en la variable de Airflow
image_name
.Usa DataprocCreateBatchOperator con tu imagen personalizada:
Usa el servicio de Dataproc Metastore con DataprocCreateBatchOperator
Para usar un servicio de Dataproc Metastore desde un DAG, haz lo siguiente:
Verifica que tu servicio de almacén de metadatos ya se haya iniciado.
Para obtener información sobre cómo iniciar un servicio de almacén de metadatos, consulta Habilita o inhabilita Dataproc Metastore.
Si deseas obtener información detallada sobre el operador de lotes para crear la configuración, consulta PeripheralsConfig.
Una vez que el servicio de almacén de metadatos esté en funcionamiento, especifica su nombre en la variable
metastore_cluster
y su región en la variable de Airflowregion_name
.Usa el servicio de almacén de metadatos en DataprocCreateBatchOperator:
DataprocDeleteBatchOperator
Puedes usar DataprocDeleteBatchOperator para borrar un lote según el ID del lote de la carga de trabajo.
DataprocListBatchesOperator
DataprocDeleteBatchOperator enumera los lotes que existen dentro de una región y un project_id determinados.
DataprocGetBatchOperator
DataprocGetBatchOperator recupera una carga de trabajo por lotes particular.