Prueba, sincroniza y, luego, implementa tus DAG desde GitHub

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

En esta guía, se explica cómo crear una canalización de CI/CD para probar, sincronizar e implementar DAG en tu entorno de Cloud Composer desde tu repositorio de GitHub.

Si solo quieres sincronizar datos de otros servicios, consulta Cómo transferir datos de otros servicios.

Descripción general de la canalización de CI/CD

La canalización de CI/CD para probar, sincronizar e implementar DAG tiene los siguientes pasos:

1. Realizas un cambio en un DAG y lo envías a una rama de desarrollo en tu repositorio.

2. Abre una solicitud de extracción en la rama principal de tu repositorio.

3. Cloud Build ejecuta pruebas de unidades para verificar que tu DAG sea válido.

4. Se aprobó tu solicitud de extracción y se combinó con la rama principal de tu repositorio.

5. Cloud Build sincroniza tu entorno de desarrollo de Cloud Composer con estos cambios nuevos.

6. Verificas que el DAG se comporte según lo esperado en tu entorno de desarrollo.

7. Si tu DAG funciona según lo previsto, súbelo a tu entorno de producción de Cloud Composer.

Objetivos

Antes de comenzar

• En esta guía, se supone que trabajas con dos entornos de Cloud Composer idénticos: un entorno de desarrollo y un entorno de producción.

  Para los fines de esta guía, solo configurarás una canalización de CI/CD para tu entorno de desarrollo. Asegúrate de que el entorno que uses no sea un entorno de producción.

• En esta guía, se supone que tienes tus DAG y sus pruebas almacenados en un repositorio de GitHub.

  La canalización de CI/CD de ejemplo muestra el contenido de un repositorio de ejemplo. Los DAGs y las pruebas se almacenan en el directorio dags/, con archivos de requisitos, el archivo de restricciones y los archivos de configuración de Cloud Build almacenados en el nivel superior. La utilidad de sincronización de DAG y sus requisitos se encuentran en el directorio utils.

Crea un trabajo de verificación previa al envío y pruebas de unidades

El primer trabajo de Cloud Build ejecuta una verificación previa al envío, que ejecuta pruebas de unidades para tus DAG.

Agrega pruebas de unidades

Si aún no lo hiciste, crea pruebas de unidades para tus DAG. Guarda estas pruebas junto con los DAG en tu repositorio, cada uno con el sufijo _test. Por ejemplo, el archivo de prueba para el DAG en example_dag.py es example_dag_test.py. Estas son las pruebas que se ejecutan como una verificación previa al envío en tu repositorio.

Crea la configuración de YAML de Cloud Build para la verificación previa al envío

En tu repositorio, crea un archivo YAML llamado test-dags.cloudbuild.yaml que configure tu trabajo de Cloud Build para las verificaciones previas al envío. En él, se indican tres pasos:

1. Instala las dependencias que necesitan tus DAGs.
2. Instala las dependencias que necesitan tus pruebas de unidades.
3. Ejecuta las pruebas del DAG.

steps:
  # install dependencies
  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements.txt", "-c", "constraints.txt", "--user"]

  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements-test.txt", "--user"]

  # run in python 3.8 which is latest version in Cloud Composer
  - name: python:3.8-slim
    entrypoint: python3.8
    args: ["-m", "pytest", "-s", "dags/"]

Crea el activador de Cloud Build para la verificación previa al envío

Sigue la guía Compilar repositorios de GitHub para crear un activador basado en una app de GitHub con las siguientes configuraciones:

• Nombre: test-dags

• Evento: Pull Request

• Fuente: Repositorio: Elige tu repositorio.

• Fuente: Rama base: ^main$ (cambia main por el nombre de la rama base de tu repositorio, si es necesario)

• Fuente: Control de comentarios: No es obligatorio

• Configuración de compilación: Archivo de configuración de Cloud Build: /test-dags.cloudbuild.yaml (la ruta de acceso a tu archivo de compilación)

Crea un trabajo de sincronización de DAG y agrega una secuencia de comandos de utilidad de DAG

A continuación, configura un trabajo de Cloud Build que ejecute una secuencia de comandos de utilidad de DAG. La secuencia de comandos de utilidad de este trabajo sincroniza tus DAGs con tu entorno de Cloud Composer después de que se combinan con la rama principal de tu repositorio.

Agrega la secuencia de comandos de utilidad de DAGs

Agrega la secuencia de comandos de utilidad del DAG a tu repositorio. Esta secuencia de comandos de utilidad copia todos los archivos DAG del directorio dags/ de tu repositorio en un directorio temporal, ignorando todos los archivos Python que no sean DAG. Luego, la secuencia de comandos usa la biblioteca cliente de Cloud Storage para subir todos los archivos de ese directorio temporal al directorio dags/ en el bucket de tu entorno de Cloud Composer.

from __future__ import annotations

import argparse
import glob
import os
from shutil import copytree, ignore_patterns
import tempfile

# Imports the Google Cloud client library
from google.cloud import storage


def _create_dags_list(dags_directory: str) -> tuple[str, list[str]]:
    temp_dir = tempfile.mkdtemp()

    # ignore non-DAG Python files
    files_to_ignore = ignore_patterns("__init__.py", "*_test.py")

    # Copy everything but the ignored files to a temp directory
    copytree(dags_directory, f"{temp_dir}/", ignore=files_to_ignore, dirs_exist_ok=True)

    # The only Python files left in our temp directory are DAG files
    # so we can exclude all non Python files
    dags = glob.glob(f"{temp_dir}/*.py")
    return (temp_dir, dags)


def upload_dags_to_composer(
    dags_directory: str, bucket_name: str, name_replacement: str = "dags/"
) -> None:
    """
    Given a directory, this function moves all DAG files from that directory
    to a temporary directory, then uploads all contents of the temporary directory
    to a given cloud storage bucket
    Args:
        dags_directory (str): a fully qualified path to a directory that contains a "dags/" subdirectory
        bucket_name (str): the GCS bucket of the Cloud Composer environment to upload DAGs to
        name_replacement (str, optional): the name of the "dags/" subdirectory that will be used when constructing the temporary directory path name Defaults to "dags/".
    """
    temp_dir, dags = _create_dags_list(dags_directory)

    if len(dags) > 0:
        # Note - the GCS client library does not currently support batch requests on uploads
        # if you have a large number of files, consider using
        # the Python subprocess module to run gsutil -m cp -r on your dags
        # See https://cloud.google.com/storage/docs/gsutil/commands/cp for more info
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)

        for dag in dags:
            # Remove path to temp dir
            dag = dag.replace(f"{temp_dir}/", name_replacement)

            try:
                # Upload to your bucket
                blob = bucket.blob(dag)
                blob.upload_from_filename(dag)
                print(f"File {dag} uploaded to {bucket_name}/{dag}.")
            except FileNotFoundError:
                current_directory = os.listdir()
                print(
                    f"{name_replacement} directory not found in {current_directory}, you may need to override the default value of name_replacement to point to a relative directory"
                )
                raise

    else:
        print("No DAGs to upload.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--dags_directory",
        help="Relative path to the source directory containing your DAGs",
    )
    parser.add_argument(
        "--dags_bucket",
        help="Name of the DAGs bucket of your Composer environment without the gs:// prefix",
    )

    args = parser.parse_args()

    upload_dags_to_composer(args.dags_directory, args.dags_bucket)

Crea la configuración YAML de Cloud Build para sincronizar DAGs

En tu repositorio, crea un archivo YAML llamado add-dags-to-composer.cloudbuild.yaml que configure tu trabajo de Cloud Build para sincronizar DAGs. El proceso consta de dos pasos:

1. Instala las dependencias que necesita la secuencia de comandos de utilidad de los DAG.

2. Ejecuta la secuencia de comandos de utilidad para sincronizar los DAGs de tu repositorio con tu entorno de Cloud Composer.

steps:
  # install dependencies
  - name: python
    entrypoint: pip
    args: ["install", "-r", "utils/requirements.txt", "--user"]

  # run
  - name: python
    entrypoint: python
    args: ["utils/add_dags_to_composer.py", "--dags_directory=${_DAGS_DIRECTORY}", "--dags_bucket=${_DAGS_BUCKET}"]

Crea el activador de Cloud Build

Sigue la guía Compilar repositorios de GitHub para crear un activador basado en una app de GitHub con las siguientes configuraciones:

• Nombre: add-dags-to-composer

• Evento: Envío a una rama

• Fuente: Repositorio: Elige tu repositorio.

• Fuente: Rama base: ^main$ (cambia main por el nombre de la rama base de tu repositorio, si es necesario)

• Fuente: Filtro de archivos incluidos (glob): dags/**

• Configuración de compilación: Archivo de configuración de Cloud Build: /add-dags-to-composer.cloudbuild.yaml (la ruta de acceso a tu archivo de compilación)

En la configuración avanzada, agrega dos variables de sustitución:

• _DAGS_DIRECTORY: Es el directorio en el que se encuentran los DAG en tu repositorio. Si usas el repositorio de ejemplo de esta guía, es dags/.

• _DAGS_BUCKET: Es el bucket de Cloud Storage que contiene el directorio dags/ en tu entorno de desarrollo de Cloud Composer. Omite el prefijo gs://. Por ejemplo: us-central1-example-env-1234ab56-bucket

Prueba tu canalización de CI/CD

En esta sección, seguirás un flujo de desarrollo de DAG que utiliza los activadores de Cloud Build que acabas de crear.

Ejecuta un trabajo de envío previo

Crea una solicitud de extracción en tu rama principal para probar la compilación. Ubica la verificación previa al envío en la página. Haz clic en Detalles y elige Ver más detalles en Google Cloud Build para ver los registros de tu compilación en la consola deGoogle Cloud .

Si falló la verificación previa a la confirmación, consulta Cómo solucionar errores de compilación.

Valida que tu DAG funcione en tu entorno de desarrollo de Cloud Composer

Después de que se apruebe tu solicitud de extracción, combínala con tu rama principal. Usa la consola deGoogle Cloud para ver los resultados de la compilación. Si tienes muchos activadores de Cloud Build, puedes filtrar tus compilaciones por el nombre del activador add-dags-to-composer.

Después de que el trabajo de sincronización de Cloud Build se complete correctamente, el DAG sincronizado aparecerá en tu entorno de desarrollo de Cloud Composer. Allí, puedes validar que el DAG funcione según lo esperado.

Agrega el DAG a tu entorno de producción

Después de que el DAG funcione según lo esperado, agrégalo manualmente a tu entorno de producción. Para ello, sube el archivo DAG al directorio dags/ en el bucket de tu entorno de producción de Cloud Composer.

Si falló tu trabajo de sincronización de DAG o si tu DAG no se comporta como esperas en tu entorno de desarrollo de Cloud Composer, consulta Cómo abordar las fallas de compilación.

Cómo abordar los errores de compilación

En esta sección, se explica cómo abordar situaciones comunes de fallas en la compilación.

¿Qué sucede si falla mi verificación previa al envío?

En tu solicitud de extracción, haz clic en Detalles y elige Ver más detalles en Google Cloud Build para ver los registros de compilación en la consola deGoogle Cloud . Usa estos registros para depurar el problema con tu DAG. Una vez que hayas resuelto los problemas, confirma la corrección y envíala a tu rama. La verificación previa a la confirmación se ejecuta de nuevo, y puedes seguir iterando con los registros como herramienta de depuración.

¿Qué sucede si falló mi trabajo de sincronización de DAG?

Usa la Google Cloud consola paraver los resultados de la compilación. Si tienes muchos activadores de Cloud Build, puedes filtrar tus compilaciones por el nombre del activador add-dags-to-composer. Examina los registros del trabajo de compilación y resuelve los errores. Si necesitas ayuda adicional para resolver los errores, utiliza los canales de asistencia.

¿Qué sucede si mi DAG no funciona correctamente en mi entorno de Cloud Composer?

Si tu DAG no funciona como se espera en tu entorno de desarrollo de Cloud Composer, no lo promuevas manualmente a tu entorno de producción de Cloud Composer. Como alternativa, toma una de las siguientes medidas:

• Revierte la solicitud de extracción con los cambios que dañaron tu DAG para restablecerlo al estado inmediatamente anterior a tus cambios (esto también revierte todos los demás archivos de esa solicitud de extracción).
• Crea una nueva solicitud de extracción para revertir manualmente los cambios en el DAG dañado.
• Crea una nueva solicitud de extracción para corregir los errores en tu DAG.

Seguir cualquiera de estos pasos activa una nueva verificación previa a la confirmación y, tras la combinación, el trabajo de sincronización del DAG.

¿Qué sigue?

• Ejecuta entornos locales de Airflow
• Escribir DAG
• Programa y activa DAG
• Prueba DAG