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

    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.

    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 operadores para describir el orden en el que se debe completar el trabajo.

    # 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:

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 y contrib.

BashOperator

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

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 los siguientes paquetes:

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.

from airflow.operators import email_operator
    # Send email confirmation
    email_summary = email_operator.EmailOperator(
        task_id='email_summary',
        to=models.Variable.get('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.

from airflow import models
default_dag_args = {
    'start_date': yesterday,
    # Email whenever an Operator in the DAG fails.
    'email': models.Variable.get('email'),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

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

Lineamientos

  1. 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.

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

  3. No definas subDAG como objetos de nivel superior.

    En general, Airflow selecciona objetos DAG en el espacio de nombres global de un módulo en el directorio dags/ como DAG de nivel superior. Los subDags definidos como objetos de nivel superior se ejecutan en sus propias programaciones además de las programaciones de otros DAG que incorporan los subDags.

  4. Coloca los archivos que se requieren en el tiempo de análisis de DAG en el directorio dags/ que no está en el directorio data/. El directorio data/ no está activado en el servidor web.

FAQs

Crea los DAG

Uso de operadores

Qué sigue