Escribe DAG (flujos de trabajo)

En esta guía, se muestra cómo escribir un grafo acíclico dirigido por Apache Airflow (DAG) que se ejecuta en un entorno de Cloud Composer.

Estructura un DAG

Un DAG de Airflow se define en un archivo de Python y está compuesto por los siguientes componentes: una definición de DAG, operadores y relaciones de operador. En los siguientes fragmentos de código, se muestran ejemplos de cada componente fuera de contexto:

  1. Una definición de DAG

    Airflow 2

    import datetime
    
    from airflow import models
    default_dag_args = {
        # The start_date describes when a DAG is valid / can be run. Set this to a
        # fixed point in time rather than dynamically, since it is evaluated every
        # time a DAG is parsed. See:
        # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
        'start_date': datetime.datetime(2018, 1, 1),
    }
    
    # Define a DAG (directed acyclic graph) of tasks.
    # Any task you create within the context manager is automatically added to the
    # DAG object.
    with models.DAG(
            'composer_sample_simple_greeting',
            schedule_interval=datetime.timedelta(days=1),
            default_args=default_dag_args) as dag:

    Airflow 1

    import datetime
    
    from airflow import models
    default_dag_args = {
        # The start_date describes when a DAG is valid / can be run. Set this to a
        # fixed point in time rather than dynamically, since it is evaluated every
        # time a DAG is parsed. See:
        # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
        'start_date': datetime.datetime(2018, 1, 1),
    }
    
    # Define a DAG (directed acyclic graph) of tasks.
    # Any task you create within the context manager is automatically added to the
    # DAG object.
    with models.DAG(
            'composer_sample_simple_greeting',
            schedule_interval=datetime.timedelta(days=1),
            default_args=default_dag_args) as dag:

  2. Operadores para describir el trabajo que se debe realizar. Una creación de instancias de un operador se denomina tarea.

    Airflow 2

    from airflow.operators import bash_operator
    from airflow.operators import python_operator
        def greeting():
            import logging
            logging.info('Hello World!')
    
        # An instance of an operator is called a task. In this case, the
        # hello_python task calls the "greeting" Python function.
        hello_python = python_operator.PythonOperator(
            task_id='hello',
            python_callable=greeting)
    
        # Likewise, the goodbye_bash task calls a Bash script.
        goodbye_bash = bash_operator.BashOperator(
            task_id='bye',
            bash_command='echo Goodbye.')

    Airflow 1

    from airflow.operators import bash_operator
    from airflow.operators import python_operator
        def greeting():
            import logging
            logging.info('Hello World!')
    
        # An instance of an operator is called a task. In this case, the
        # hello_python task calls the "greeting" Python function.
        hello_python = python_operator.PythonOperator(
            task_id='hello',
            python_callable=greeting)
    
        # Likewise, the goodbye_bash task calls a Bash script.
        goodbye_bash = bash_operator.BashOperator(
            task_id='bye',
            bash_command='echo Goodbye.')

  3. Relaciones entre tareas para describir el orden en el que se debe completar el trabajo.

    Airflow 2

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

    Airflow 1

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

El siguiente flujo de trabajo es un ejemplo de trabajo completo y consta de dos tareas: una tarea hello_python y una goodbye_bash:

Airflow 2

from __future__ import print_function

import datetime

from airflow import models
from airflow.operators import bash_operator
from airflow.operators import python_operator

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    'start_date': datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
        'composer_sample_simple_greeting',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:
    def greeting():
        import logging
        logging.info('Hello World!')

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id='hello',
        python_callable=greeting)

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id='bye',
        bash_command='echo Goodbye.')

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

Airflow 1

from __future__ import print_function

import datetime

from airflow import models
from airflow.operators import bash_operator
from airflow.operators import python_operator

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    'start_date': datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
        'composer_sample_simple_greeting',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:
    def greeting():
        import logging
        logging.info('Hello World!')

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id='hello',
        python_callable=greeting)

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id='bye',
        bash_command='echo Goodbye.')

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

Consulta el instructivo de Airflow y los conceptos de Airflow para obtener más información sobre cómo definir los DAG de Airflow.

Operadores

En los siguientes ejemplos, se muestran algunos operadores populares de Airflow. Para obtener una referencia fidedigna de los operadores de Airflow, consulta la Referencia de la API de Apache Airflow o explora el código fuente de los operadores core, contrib y providers.

BashOperator

Usa BashOperator para ejecutar programas de línea de comandos.

Airflow 2

from airflow.operators import bash

    # Create BigQuery output dataset.
    make_bq_dataset = bash.BashOperator(
        task_id="make_bq_dataset",
        # Executing 'bq' command requires Google Cloud SDK which comes
        # preinstalled in Cloud Composer.
        bash_command=f"bq ls {bq_dataset_name} || bq mk {bq_dataset_name}",
    )

Airflow 1

from airflow.operators import bash_operator

    # Create BigQuery output dataset.
    make_bq_dataset = bash_operator.BashOperator(
        task_id="make_bq_dataset",
        # Executing 'bq' command requires Google Cloud SDK which comes
        # preinstalled in Cloud Composer.
        bash_command="bq ls {} || bq mk {}".format(bq_dataset_name, bq_dataset_name),
    )

Cloud Composer ejecuta los comandos proporcionados en una secuencia de comandos de Bash en un trabajador. El trabajador es un contenedor de Docker basado en Debian que incluye varios paquetes.

PythonOperator

Usa PythonOperator para ejecutar código de Python arbitrario.

Cloud Composer ejecuta el código de Python en un contenedor que incluye paquetes para la versión de la imagen de Cloud Composer que se usa en tu entorno.

Para instalar paquetes adicionales de Python, consulta cómo instalar dependencias de Python.

Operadores de Google Cloud

Usa los operadores de Airflow de Google Cloud para ejecutar tareas que usen productos de Google Cloud. Cloud Composer configura automáticamente una conexión de Airflow en el proyecto del entorno.

EmailOperator

Usa el EmailOperator para enviar correos electrónicos desde un DAG. Para enviar correos electrónicos desde un entorno de Cloud Composer, debes configurar tu entorno para usar SendGrid.

Airflow 2

from airflow.operators import email

    # Send email confirmation (you will need to set up the email operator
    # See https://cloud.google.com/composer/docs/how-to/managing/creating#notification
    # for more info on configuring the email operator in Cloud Composer)
    email_summary = email.EmailOperator(
        task_id="email_summary",
        to="{{var.value.email}}",
        subject="Sample BigQuery notify data ready",
        html_content="""
        Analyzed Stack Overflow posts data from {min_date} 12AM to {max_date}
        12AM. The most popular question was '{question_title}' with
        {view_count} views. Top 100 questions asked are now available at:
        {export_location}.
        """.format(
            min_date=min_query_date,
            max_date=max_query_date,
            question_title=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][0] }}"
            ),
            view_count=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][1] }}"
            ),
            export_location=output_file,
        ),
    )

Airflow 1

from airflow.operators import email_operator

    # Send email confirmation
    email_summary = email_operator.EmailOperator(
        task_id="email_summary",
        to="{{var.value.email}}",
        subject="Sample BigQuery notify data ready",
        html_content="""
        Analyzed Stack Overflow posts data from {min_date} 12AM to {max_date}
        12AM. The most popular question was '{question_title}' with
        {view_count} views. Top 100 questions asked are now available at:
        {export_location}.
        """.format(
            min_date=min_query_date,
            max_date=max_query_date,
            question_title=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][0] }}"
            ),
            view_count=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][1] }}"
            ),
            export_location=output_file,
        ),
    )

Notificaciones

Configura email_on_failure como True para enviar una notificación por correo electrónico cuando un operador del DAG falle. Para enviar notificaciones por correo electrónico desde un entorno de Cloud Composer, debes configurar tu entorno para usar SendGrid.

Airflow 2

from airflow import models

default_dag_args = {
    "start_date": yesterday,
    # Email whenever an Operator in the DAG fails.
    "email": "{{var.value.email}}",
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "project_id": project_id,
}

with models.DAG(
    "composer_sample_bq_notify",
    schedule_interval=datetime.timedelta(weeks=4),
    default_args=default_dag_args,
) as dag:

Airflow 1

from airflow import models

default_dag_args = {
    "start_date": yesterday,
    # Email whenever an Operator in the DAG fails.
    "email": "{{var.value.email}}",
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "project_id": "{{var.value.gcp_project}}",
}

with models.DAG(
    "composer_sample_bq_notify",
    schedule_interval=datetime.timedelta(weeks=4),
    default_args=default_dag_args,
) as dag:

Lineamientos

  1. Usa Airflow 2 en lugar de Airflow 1.

    La comunidad de Airflow ya no publica versiones secundarias o de parches nuevas para Airflow 1.

  2. Coloca cualquier biblioteca de Python personalizada en el archivo ZIP de un DAG en un directorio anidado. No coloques bibliotecas en el nivel superior del directorio de DAG.

    Cuando Airflow escanea la carpeta dags/, Airflow solo busca DAG en los módulos de Python que se encuentran en el nivel superior de la carpeta de DAG y en el nivel superior de un archivo ZIP ubicado también en la carpeta dags/ de nivel superior. Si Airflow encuentra un módulo de Python en un archivo ZIP que no contiene las substrings airflow y DAG, Airflow deja de procesar el archivo ZIP. Airflow solo muestra los DAG encontrados hasta ese momento.

  3. Para la tolerancia a errores, no definas varios objetos DAG en el mismo módulo de Python.

  4. No uses SubDAG. En su lugar, usa alternativas.

  5. Coloca los archivos que se requieren en el tiempo de análisis del DAG en la carpeta dags/, no en el directorio data/.

  6. Implementa pruebas de unidades para tus DAG

  7. Prueba los DAG desarrollados o modificados como se recomienda en las instrucciones para probar DAG

  8. Verifica que los DAG desarrollados no aumenten demasiado los tiempos de análisis de DAG.

  9. Las tareas de Airflow pueden fallar por varios motivos. Para evitar fallas en las ejecuciones de DAG, recomendamos habilitar los reintentos de tareas. Establecer la cantidad máxima de reintentos en 0 significa que no se realizan más reintentos.

    Recomendamos anular la opción default_task_retries con un valor para la tarea que se retire aparte de 0. Además, puedes configurar el parámetro de reintentos a nivel de tarea (si es necesario).

  10. Si deseas usar GPU en tus tareas de Airflow, crea un clúster de GKE independiente basado en nodos que usen máquinas con GPU. Usa GKEStartPodOperator para ejecutar tus tareas.

  11. Evita ejecutar tareas pesadas de CPU y memoria en el grupo de nodos del clúster en el que se ejecutan otros componentes de Airflow (programadores, trabajadores y servidores web). En su lugar, usa KubernetesPodOperator o GKEPodOperators. Seleccionar

  12. Cuando implementes DAG en un entorno, sube solo los archivos que sean absolutamente necesarios para interpretar y ejecutar DAG en la carpeta /dags.

  13. Limita la cantidad de archivos DAG en la carpeta /dags

    Airflow analiza continuamente los DAG en la carpeta /dags. El análisis es un proceso que recorre la carpeta DAGs y la cantidad de archivos que se deben cargar (con sus dependencias) afecta el rendimiento del análisis del DAG y la programación de tareas. Es mucho más eficiente usar 100 archivos con 100 DAG cada uno que 10,000 archivos con 1 DAG, por lo que se recomienda esta optimización. Esta optimización es el equilibrio entre el tiempo de análisis y la eficiencia de la creación y la administración de DAG.

    También puedes considerar, p.ej., para implementar 10,000 archivos DAG, puedes crear 100 archivos ZIP cada uno con 100 archivos DAG.

    Además de las sugerencias anteriores, si tienes más de 10,000 archivos DAG, la generación de DAG de manera programática podría ser una buena opción. Por ejemplo, puedes implementar un solo archivo DAG de Python que genera cierta cantidad de objetos DAG (p.ej., 20, 100 objetos DAG).

Preguntas frecuentes sobre la escritura de DAG

¿Cómo minimizo la repetición de código si quiero ejecutar las mismas tareas o tareas similares en varios DAG?

Te recomendamos definir bibliotecas y wrappers para minimizar la repetición de código.

¿Cómo vuelvo a usar el código entre los archivos DAG?

Coloca tus funciones de utilidad en una biblioteca local de Python y, luego, importa las funciones. Puedes hacer referencia a las funciones en cualquier DAG ubicado en la carpeta dags/ del depósito de tu entorno.

¿Cómo minimizo el riesgo de que surjan definiciones diferentes?

Por ejemplo, tienes dos equipos que desean agregar datos sin procesar a las métricas de ingresos. Los equipos escriben dos tareas con pequeñas diferencias entre sí que logran lo mismo. Se deben definir bibliotecas para trabajar con los datos de ingresos, de modo que los implementadores de DAG aclaren la definición de ingresos que se agrega.

¿Cómo configuro las dependencias entre los DAG?

Esto dependerá de cómo deseas definir la dependencia.

Si tienes dos DAG (DAG A y DAG B) y quieres que el DAG B se active después del DAG A, puedes poner un TriggerDagRunOperator al final del Dag A.

Si el DAG B solo depende de un artefacto generado por el DAG A, como un mensaje de Pub/Sub, es posible que un sensor funcione mejor.

Si el DAG B se integra perfectamente al DAG A, es posible que puedas combinar los dos DAG en uno.

¿Cómo transfiero los ID de ejecución únicos a un DAG y sus tareas?

Por ejemplo, deseas transmitir nombres de clústeres de Dataproc y rutas de archivos.

Puedes generar un ID único aleatorio si muestras str(uuid.uuid4()) en un PythonOperator. Esto coloca el ID en XComs para que puedas hacer referencia al ID en otros operadores a través de campos de plantilla.

Antes de generar un uuid, considera si un ID específico de DagRun sería más valioso. También puedes hacer referencia a estos ID en sustituciones de Jinja mediante macros.

¿Cómo separo las tareas de un DAG?

Cada tarea debe ser una unidad de trabajo idempotente. Por lo tanto, debes evitar encapsular un flujo de trabajo de varios pasos en una sola tarea, como un programa complejo que se ejecute en un PythonOperator.

¿Debo definir varias tareas en un solo DAG para agregar datos de varias fuentes?

Por ejemplo, tienes varias tablas con datos sin procesar y quieres crear agregados diarios para cada una. Las tareas no dependen una de otra. ¿Debes crear una tarea y un DAG para cada tabla o crear un DAG general?

Si estás de acuerdo con que cada tarea comparta las mismas propiedades a nivel del DAG, como schedule_interval, entonces tiene sentido definir varias tareas en un solo DAG. De lo contrario, para minimizar la repetición de código, se pueden generar varios DAG a partir de un único módulo de Python colocándolos en globals() del módulo.

¿Cómo puedo limitar la cantidad de tareas simultáneas que se ejecutan en un DAG?

Por ejemplo, quieres evitar superar las cuotas o los límites de uso de la API o evitar ejecutar demasiados procesos simultáneos.

Puedes definir grupos de Airflow en la IU web de Airflow y asociar tareas con grupos existentes en tus DAG.

Preguntas frecuentes sobre el uso de operadores

¿Debo usar DockerOperator?

No recomendamos usar DockerOperator, a menos que se use para iniciar contenedores en una instalación remota de Docker (no dentro de un clúster del entorno). En un entorno de Cloud Composer, el operador no tiene acceso a los daemons de Docker.

En su lugar, usa KubernetesPodOperator o GKEStartPodOperator. Estos operadores inician pods de Kubernetes en clústeres de Kubernetes o de GKE, respectivamente. Tenga en cuenta que no recomendamos iniciar Pods en el clúster de un entorno, ya que esto puede generar competencia en los recursos.

¿Debo usar SubDagOperator?

No recomendamos usar SubDagOperator.

Usa las alternativas que se sugieren en las instrucciones para agrupar tareas

¿Debería ejecutar código de Python solo en PythonOperators para separar completamente los operadores de Python?

Según tu objetivo, tienes algunas opciones.

Si tu única preocupación es mantener dependencias de Python separadas, puedes usar PythonVirtualenvOperator.

Considera usar KubernetesPodOperator. Este operador te permite definir pods de Kubernetes y ejecutar los pods en otros clústeres.

¿Cómo agrego paquetes binarios personalizados o que no son de PyPI?

Puedes instalar paquetes alojados en repositorios de paquetes privados.

También puedes usar KubernetesPodOperator para ejecutar un pod de Kubernetes con tu propia imagen compilada con paquetes personalizados.

¿Cómo transmito uniformemente los argumentos a un DAG y sus tareas?

Puedes usar la compatibilidad integrada de Airflow para las plantillas de Jinja a fin de pasar argumentos que se pueden usar en campos de plantilla.

¿Cuándo ocurre la sustitución de plantilla?

La sustitución de plantilla ocurre en los trabajadores de Airflow justo antes de que se llame a la función pre_execute de un operador. En la práctica, esto significa que las plantillas no se sustituyen hasta justo antes de que se ejecute una tarea.

¿Cómo puedo saber qué argumentos del operador admiten la sustitución de plantillas?

Los argumentos de operadores que admiten la sustitución de plantillas de Jinja2 se marcan de forma explícita como tales.

Busca el campo template_fields en la definición de operador, que contiene una lista de nombres de argumentos que se someterán a la sustitución de plantillas.

Por ejemplo, consulta BashOperator, que admite plantillas para los argumentos bash_command y env.

Evita usar operadores de Airflow obsoletos

Los operadores enumerados en la siguiente tabla ya no están disponibles. Evite usarlas en sus DAG. En su lugar, usa las alternativas actualizadas.

Operador obsoleto Operador que se usará
BigQueryExecuteQueryOperator BigQueryInsertJobOperator
BigQueryPatchDatasetOperator BigQueryUpdateTableOperator
DataflowCreateJavaJobOperator BeamRunJavaPipelineOperator
DataflowCreatePythonJobOperator BeamRunPythonPipelineOperator
OperadorDataprocClusterCluster OperadorClusterClusterUpdate
DataprocSubmitPigJobOperator DataprocSubmitJobOperator
DataprocSubmitSparkSqlJobOperator DataprocSubmitJobOperator
DataprocSubmitSparkJobOperator DataprocSubmitJobOperator
DataprocSubmitHadoopJobOperator DataprocSubmitJobOperator
DataprocSubmitPySparkJobOperator DataprocSubmitJobOperator
MLEngineManageModelOperator MLEngineCreateModelOperator, MLEngineGetModelOperator
MLEngineManageVersionOperator MLEngineCreateVersion, MLEngineSetDefaultVersion, MLEngineListVersions, MLEngineDeleteVersion
GCSObjectsWtihPrefixExistenceSensor GCSObjectsWithPrefixExistenceSensor

¿Qué sigue?