Probar DAGs de Airflow

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Antes de implementar DAGs en producción, puedes ejecutar subcomandos de la CLI de Airflow para analizar el código de los DAGs en el mismo contexto en el que se ejecutan.

Pruebas durante la creación de DAGs

Puedes ejecutar una sola instancia de tarea de forma local y ver el resultado del registro. Al ver el resultado, puedes comprobar si hay errores de sintaxis y de tareas. Las pruebas locales no comprueban las dependencias ni comunican el estado a la base de datos.

Te recomendamos que pongas los DAGs en una carpeta data/test de tu entorno de prueba.

Crear un directorio de prueba

En el segmento de tu entorno, crea un directorio de prueba y copia tus DAGs en él.

gcloud storage cp BUCKET_NAME/dags \
  BUCKET_NAME/data/test --recursive

Haz los cambios siguientes:

  • BUCKET_NAME: el nombre del bucket asociado a tu entorno de Cloud Composer.

Ejemplo:

gcloud storage cp gs://us-central1-example-environment-a12bc345-bucket/dags \
  gs://us-central1-example-environment-a12bc345-bucket/data/test --recursive

Para obtener más información sobre cómo subir DAGs, consulta el artículo Añadir y actualizar DAGs.

Comprobar si hay errores de sintaxis

Para comprobar si hay errores de sintaxis en los DAGs que has subido a la carpeta /data/test, introduce el siguiente comando gcloud:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  dags list -- --subdir /home/airflow/gcs/data/test

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  list_dags -- -sd /home/airflow/gcs/data/test

Haz los cambios siguientes:

  • ENVIRONMENT_NAME: el nombre del entorno.
  • ENVIRONMENT_LOCATION: la región en la que se encuentra el entorno.

Comprobar si hay errores en las tareas

Para comprobar si hay errores específicos de tareas en los DAGs que has subido a la carpeta /data/test, ejecuta el siguiente comando gcloud:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  DAG_ID TASK_ID \
  DAG_EXECUTION_DATE

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  test -- -sd /home/airflow/gcs/data/test DAG_ID \
  TASK_ID DAG_EXECUTION_DATE

Haz los cambios siguientes:

  • ENVIRONMENT_NAME: el nombre del entorno.
  • ENVIRONMENT_LOCATION: la región en la que se encuentra el entorno.
  • DAG_ID: el ID del DAG.
  • TASK_ID: el ID de la tarea.
  • DAG_EXECUTION_DATE: la fecha de ejecución del DAG. Esta fecha se usa con fines de creación de plantillas. Independientemente de la fecha que especifique aquí, el DAG se ejecuta inmediatamente.

Ejemplo:

Airflow 2 

gcloud composer environments run \
  example-environment \
  --location us-central1 \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  hello_world print_date 2021-04-22

Airflow 1 

gcloud composer environments run example-environment \
  --location us-central1 \
  test -- -sd /home/airflow/gcs/data/test \
  hello_world print_date 2021-04-22

Actualizar y probar un DAG implementado

Para probar las actualizaciones de tus DAGs en tu entorno de prueba, sigue estos pasos:

  1. Copia el DAG implementado que quieras actualizar en data/test.
  2. Actualiza el DAG.
  3. Prueba el DAG.
    1. Comprueba si hay errores de sintaxis.
    2. Comprueba si hay errores específicos de la tarea.
  4. Asegúrate de que el DAG se ejecute correctamente.
  5. Desactiva el DAG en tu entorno de prueba.
    1. Ve a la interfaz de usuario de Airflow > página DAGs.
    2. Si el DAG que vas a modificar se ejecuta constantemente, desactívalo.
    3. Para agilizar las tareas pendientes, haz clic en la tarea y en Marcar como completada.
  6. Despliega el DAG en tu entorno de producción.
    1. Desactiva el DAG en tu entorno de producción.
    2. Sube el DAG actualizado a la carpeta dags/ de tu entorno de producción.

Preguntas frecuentes sobre las pruebas de DAGs

¿Cómo puedo aislar las ejecuciones de DAG en mis entornos de producción y de prueba?

Por ejemplo, Airflow tiene un repositorio global de código fuente en la carpeta dags/ que comparten todas las ejecuciones de DAG. Quieres actualizar el código fuente en producción o hacer pruebas sin interferir en los DAGs en ejecución.

Airflow no proporciona un aislamiento sólido de los DAGs. Te recomendamos que mantengas entornos de producción y de prueba de Cloud Composer independientes para evitar que los DAGs de prueba interfieran con los DAGs de producción.

¿Cómo evito las interferencias de DAG al ejecutar pruebas de integración desde diferentes ramas de GitHub?

Usa nombres de tareas únicos para evitar interferencias. Por ejemplo, puedes añadir el prefijo del nombre de la rama a los IDs de tus tareas.

¿Cuál es la práctica recomendada para las pruebas de integración con Airflow?

Te recomendamos que utilices un entorno específico para las pruebas de integración con Airflow. Una forma de indicar que la ejecución del DAG se ha completado correctamente es escribir en un archivo de una carpeta de Cloud Storage y, a continuación, comprobar el contenido en tus propios casos de prueba de integración.

¿Cómo puedo colaborar de forma eficiente con otros colaboradores de DAGs?

Cada colaborador puede tener un subdirectorio en la carpeta data/ para el desarrollo.

El programador de Airflow o el servidor web no detectan automáticamente los DAGs que se añaden a la carpeta data/.

Los colaboradores de DAG pueden crear ejecuciones de DAG manuales mediante el comando gcloud composer environments run y el subcomando test con la marca --subdir para especificar el directorio de desarrollo del colaborador.

Por ejemplo:

Airflow 2 

gcloud composer environments run test-environment-name \
  tasks test -- dag-id task-id execution-date \
  --subdir /home/airflow/gcs/data/alice_dev

Airflow 1 

gcloud composer environments run test-environment-name \
  test -- dag-id task-id execution-date \
  --subdir /home/airflow/gcs/data/alice_dev

¿Cómo puedo mantener sincronizados mis entornos de implementación y producción?

Para gestionar el acceso, sigue estos pasos:

Para desplegar de desarrollo a producción, sigue estos pasos:

  • Asegúrate de que la configuración sea coherente, como las variables de entorno y los paquetes de PyPI.

  • Asegúrate de que los argumentos de los DAG sean coherentes. Para evitar la codificación, te recomendamos que uses macros y variables de Airflow.

    Por ejemplo:

    Airflow 2 

    gcloud composer environments run test-environment-name \
  variables set -- DATA_ENDPOINT_KEY DATA_ENDPOINT_VALUE

    Airflow 1 

    gcloud composer environments run test-environment-name \
  variables -- --set DATA_ENDPOINT_KEY DATA_ENDPOINT_VALUE

Siguientes pasos