Agregar y actualizar DAG

Organízate con las colecciones Guarda y clasifica el contenido según tus preferencias.

Cloud Composer 1 | Cloud Composer 2

En esta página, se describe cómo administrar los DAG en tu entorno de Cloud Composer.

Cloud Composer usa un depósito de Cloud Storage para almacenar los DAG de tu entorno de Cloud Composer. Tu entorno sincroniza los DAG de este depósito con los componentes de Airflow, como los trabajadores y programadores de Airflow.

Antes de comenzar

  • Debido a que Apache Airflow no proporciona un aislamiento eficaz del DAG, recomendamos que mantengas entornos de prueba y producción separados para evitar la interferencia del DAG. Para obtener más información, consulta Probar DAG.
  • Asegúrate de que tu cuenta tenga los permisos necesarios para administrar los DAG.
  • Los cambios en el DAG se propagan a Airflow en 3 a 5 minutos. Puedes ver el estado de la tarea en la interfaz web de Airflow.

Accede al bucket de tu entorno

Para acceder al bucket asociado a tu entorno, haz lo siguiente:

Console

  1. En Google Cloud Console, ve a la página Entornos.

    Ir a Entornos

  2. En la lista de entornos, busca una fila con el nombre de tu entorno y, en la columna Carpeta de DAG, haz clic en el vínculo DAG. Se abrirá la página Detalles del bucket . Muestra contenido de la carpeta /dags en el bucket de tu entorno.

gcloud

La CLI de gcloud tiene comandos independientes para agregar y borrar DAG en el bucket de tu entorno.

Si deseas interactuar con el depósito de tu entorno, también puedes usar la herramienta de línea de comandos de gsutil. Para obtener la dirección del bucket de tu entorno, ejecuta el siguiente comando de la CLI de gcloud:

gcloud composer environments describe ENVIRONMENT_NAME \
    --location LOCATION \
    --format="get(config.dagGcsPrefix)"

Reemplaza lo siguiente:

  • ENVIRONMENT_NAME por el nombre del entorno.
  • LOCATION por la región en la que se encuentra el entorno

Ejemplo:

gcloud beta composer environments describe example-environment \
    --location us-central1 \
    --format="get(config.dagGcsPrefix)"

API

Realiza una solicitud a la API de environments.get. En el recurso Entorno, en el recurso Entorno de configuración, en el recurso dagGcsPrefix es la dirección del depósito de tu entorno.

Ejemplo:

GET https://composer.googleapis.com/v1/projects/example-project/
locations/us-central1/environments/example-environment

Python

Usa la biblioteca google-auth a fin de obtener credenciales y la biblioteca requests para llamar a la API de REST.

import google.auth
import google.auth.transport.requests

# Authenticate with Google Cloud.
# See: https://cloud.google.com/docs/authentication/getting-started
credentials, _ = google.auth.default(
    scopes=['https://www.googleapis.com/auth/cloud-platform'])
authed_session = google.auth.transport.requests.AuthorizedSession(
    credentials)

# project_id = 'YOUR_PROJECT_ID'
# location = 'us-central1'
# composer_environment = 'YOUR_COMPOSER_ENVIRONMENT_NAME'

environment_url = (
    'https://composer.googleapis.com/v1beta1/projects/{}/locations/{}'
    '/environments/{}').format(project_id, location, composer_environment)
response = authed_session.request('GET', environment_url)
environment_data = response.json()

# Print the bucket name from the response body.
print(environment_data['config']['dagGcsPrefix'])

Agrega o actualiza un DAG

Para agregar o actualizar un DAG, mueve el archivo .py de Python del DAG a la carpeta /dags en el depósito del entorno.

Console

  1. En Google Cloud Console, ve a la página Entornos.

    Ir a Entornos

  2. En la lista de entornos, busca una fila con el nombre de tu entorno y, en la columna Carpeta de DAG, haz clic en el vínculo DAG. Se abrirá la página Detalles del bucket . Muestra contenido de la carpeta /dags en el bucket de tu entorno.

  3. Haga clic en Subir archivos. Luego, selecciona el archivo .py de Python para el DAG mediante el diálogo del navegador y confirma.

gcloud

gcloud composer environments storage dags import \
    --environment ENVIRONMENT_NAME \
    --location LOCATION \
    --source="LOCAL_FILE_TO_UPLOAD"

Reemplaza lo siguiente:

  • ENVIRONMENT_NAME por el nombre del entorno.
  • LOCATION por la región en la que se encuentra el entorno
  • LOCAL_FILE_TO_UPLOAD es el archivo .py de Python para el DAG.

Ejemplo:

gcloud beta composer environments storage dags import \
    --environment example-environment \
    --location us-central1 \
    --source="example_dag.py"

Actualiza un DAG que tenga ejecuciones activas del DAG

Si actualizas un DAG que tiene ejecuciones activas del DAG:

  • Todas las tareas que se ejecutan actualmente finalizan mediante el archivo DAG original.
  • Todas las tareas que están programadas, pero que en este momento no se están ejecutando, usan el archivo DAG actualizado.
  • Todas las tareas que ya no están presentes en el archivo DAG actualizado se marcan como eliminadas.

Actualiza los DAG que se ejecutan con frecuencia

Después de subir un archivo DAG, puede llevar algún tiempo que Airflow cargue este archivo y actualice el DAG. Si tu DAG se ejecuta con frecuencia, es posible que desees asegurarte de que use la versión actualizada del archivo de DAG. Para ello, deberás hacer lo siguiente:

  1. Pausa el DAG en la IU de Airflow.
  2. Sube un archivo DAG actualizado.
  3. Espera hasta ver las actualizaciones en la IU de Airflow. Esto significa que el programador analizó correctamente el DAG y lo actualizó en la base de datos de Airflow.

    Si la IU de Airflow muestra los DAG actualizados, esto no garantiza que los trabajadores de Airflow tengan la versión actualizada del archivo de DAG. Esto sucede porque los archivos de DAG se sincronizan de forma independiente para los programadores y trabajadores.

  4. Es posible que desees extender el tiempo de espera para asegurarte de que el archivo DAG esté sincronizado con todos los trabajadores de tu entorno. La sincronización se realiza varias veces por minuto. En un entorno en buen estado, esperar entre 20 y 30 segundos es suficiente para que todos los trabajadores se sincronicen.

  5. Si deseas asegurarte de que todos los trabajadores tengan la versión nueva del archivo de DAG, inspecciona los registros de cada trabajador individual (opcional). Para ello, deberás hacer lo siguiente:

    1. Abre la pestaña Registros de tu entorno en Google Cloud Console.

    2. Ve a Registros de Composer > Infraestructura > Sincronización de Cloud Storage y, luego, inspecciona los registros de cada trabajador de tu entorno. Busca el elemento de registro Syncing dags directory más reciente que tenga una marca de tiempo después de subir el archivo DAG nuevo. Si ves un elemento Finished syncing que lo sigue, los DAG se sincronizan con éxito en este trabajador.

  6. Reanuda el DAG.

Borra un DAG

En esta sección, se describe cómo borrar un DAG.

Borra un DAG en tu entorno

Para borrar un DAG, quita el archivo .py de Python para el DAG de la carpeta /dags del entorno en el bucket de tu entorno.

Console

  1. En Google Cloud Console, ve a la página Entornos.

    Ir a Entornos

  2. En la lista de entornos, busca una fila con el nombre de tu entorno y, en la columna Carpeta de DAG, haz clic en el vínculo DAG. Se abrirá la página Detalles del bucket . Muestra el contenido de la carpeta /dags en el depósito de tu entorno.

  3. Selecciona el archivo DAG, haz clic en Borrar y confirma la operación.

gcloud

gcloud composer environments storage dags delete \
    --environment ENVIRONMENT_NAME \
    --location LOCATION \
    DAG_FILE

Reemplaza lo siguiente:

  • ENVIRONMENT_NAME por el nombre del entorno.
  • LOCATION por la región en la que se encuentra el entorno
  • DAG_FILE por el archivo .py de Python para el DAG

Ejemplo:

gcloud beta composer environments storage dags import \
    --environment example-environment \
    --location us-central1 \
    example_dag.py

Quita un DAG de la IU de Airflow

Puedes quitar un DAG de la IU de Airflow en Airflow 1.10.0 o versiones posteriores.

Para quitar los metadatos de un DAG de la interfaz web de Airflow, sigue estos pasos:

IU de Airflow

  1. Ve a la IU de Airflow de tu entorno.
  2. Para el DAG, haz clic en Borrar DAG.

gcloud

En las versiones de Airflow 1 anteriores a 1.14.0, ejecuta el siguiente comando en la CLI de gcloud:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    delete_dag -- DAG_NAME

En Airflow 2, Airflow 1.14.0 y versiones posteriores, ejecuta el siguiente comando en la CLI de gcloud:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags delete -- DAG_NAME

Reemplaza lo siguiente:

  • ENVIRONMENT_NAME por el nombre del entorno.
  • LOCATION por la región en la que se encuentra el entorno
  • DAG_NAME es el nombre del DAG que se debe borrar.

¿Qué sigue?