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:
Una definición de DAG
Airflow 2
Airflow 1
Operadores para describir el trabajo que se debe realizar. Una creación de instancias de un operador se denomina tarea.
Airflow 2
Airflow 1
Relaciones entre tareas para describir el orden en el que se debe completar el trabajo.
Airflow 2
Airflow 1
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
Airflow 1
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
Airflow 1
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 imagen de Cloud Composer usada 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.
Los operadores de BigQuery consultan y procesan datos en BigQuery.
Airflow 2
Airflow 1
Los operadores de Dataflow ejecutan trabajos de Apache Beam en Dataflow.
Los operadores de Cloud Data Fusion administran y ejecutan canalizaciones de Cloud Data Fusion.
Los operadores de Dataproc ejecutan trabajos de Hadoop y Spark en Dataproc.
Los operadores de Datastore leen y escriben datos en Datastore.
Los operadores de AI Platform ejecutan trabajos de entrenamiento y predicción en AI Platform.
Los operadores de Cloud Storage leen y escriben datos en Cloud Storage.
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
Airflow 1
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
Airflow 1
Lineamientos
Usa Airflow 2 en lugar de Airflow 1.
La comunidad de Airflow ya no publica nuevas actualizaciones de versiones secundarias o del parche para Airflow 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 carpetadags/
de nivel superior. Si Airflow encuentra un módulo de Python en un archivo ZIP que no contiene las substringsairflow
yDAG
, Airflow deja de procesar el archivo ZIP. Airflow solo muestra los DAG encontrados hasta ese momento.Para la tolerancia a errores, no definas varios objetos DAG en el mismo módulo de Python.
No uses SubDAG. En su lugar, usa alternativas.
Coloca los archivos que se requieren en el momento de análisis del DAG en la carpeta
dags/
, no en el directoriodata/
.Prueba los DAG desarrollados o modificados, como se recomienda en las instrucciones para probar los DAG.
Verifica que los DAG desarrollados no aumenten demasiado los tiempos de análisis de DAG.
Las tareas de Airflow pueden fallar por diversos motivos. Para evitar fallas en las ejecuciones de DAG completas, recomendamos habilitar los reintentos de tareas. Configurar la cantidad máxima de reintentos en
0
implica que no se realizan reintentos.Te recomendamos anular la opción
default_task_retries
con un valor para la tarea que se quite que no sea0
. Además, puedes configurar el parámetro de reintentos a nivel de la tarea (si es necesario).Si deseas usar la GPU en tus tareas de Airflow, crea un clúster de GKE separado basado en nodos que usen máquinas con GPU. Usa GKEStartPodOperator para ejecutar tus tareas.
Evita ejecutar tareas con mucha 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
Cuando implementes DAG en un entorno, sube solo los archivos que sean absolutamente necesarios para interpretar y ejecutar DAG en la carpeta
/dags
.Limita la cantidad de archivos DAG en la carpeta
/dags
Airflow analiza los DAG de forma continua en la carpeta
/dags
. El análisis es un proceso que se repite indefinidamente a través de la carpeta de DAG y la cantidad de archivos que se deben cargar (con sus dependencias) afecta el rendimiento del análisis de 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 cada uno, por lo que se recomienda esta optimización. Esta optimización es un equilibrio entre el tiempo de análisis y la eficiencia de la creación y administración de DAG.También puedes considerar, p.ej., implementar 10,000 archivos de DAG y crear 100 archivos ZIP, cada uno con 100 archivos de DAG.
Además de las sugerencias anteriores, si tienes más de 10,000 archivos DAG, generar una DAG de manera programática puede ser una buena opción. Por ejemplo, puedes implementar un solo archivo DAG de Python que genere algún número de objetos DAG (p.ej., 20, 100 objetos DAG).
Preguntas frecuentes para escribir DAG
¿Cómo minimizo la repetición de código si quiero ejecutar las mismas tareas o tareas similares en varios DAG?
Sugerimos 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 las 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 bucket 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 puedo separar 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 de 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 GKE, respectivamente. Tenga en cuenta que no recomendamos lanzar Pods en un clúster de entorno, ya que esto puede causar la competencia de recursos.
¿Debo usar SubDagOperator
?
No recomendamos usar SubDagOperator
.
Usa alternativas como se sugiere en 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 independientes de Python, puedes usar PythonVirtualenvOperator
.
Considera usar la KubernetesPodOperator
.
Este operador le 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 la 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 realizar plantillas de Jinja a fin de pasar argumentos que se pueden usar en campos de plantillas.
¿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 del operador que admiten la sustitución de plantillas de Jinja2 se marcan de manera 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 que se enumeran en la siguiente tabla están obsoletos. Evita usarlos en tus DAG. En su lugar, use las alternativas actualizadas proporcionadas.
Operador obsoleto | Operador que se usará |
---|---|
BigQueryExecuteQueryOperator | BigQueryInsertJobOperator |
BigQueryPatchDatasetOperator | BigQueryUpdateTableOperator |
DataflowCreateJavaJobOperator | BeamRunJavaPipelineOperator |
DataflowCreatePythonJobOperator | BeamRunPythonPipelineOperator |
DataprocScaleClusterOperator | DataprocUpdateClusterOperator |
DataprocSubmitPigJobOperator | DataprocSubmitJobOperator |
DataprocSubmitSparkSqlJobOperator | DataprocSubmitJobOperator |
DataprocSubmitSparkJobOperator | DataprocSubmitJobOperator |
DataprocSubmitHadoopJobOperator | DataprocSubmitJobOperator |
DataprocSubmitPySparkJobOperator | DataprocSubmitJobOperator |
MLEngineManageModelOperator | MLEngineCreateModelOperator, MLEngineGetModelOperator |
MLEngineManageVersionOperator | MLEngineCreateVersion, MLEngineSetDefaultVersion, MLEngineListVersions, MLEngineDeleteVersion |
GCSObjectsWtihPrefixexistentceSensor | GCSObjectsWithPrefixExistenceSensor |
¿Qué sigue?
- Soluciona problemas de los DAG
- Solución de problemas del programador
- Operadores de Google
- Operadores de Google Cloud Platform
- Instructivo de Apache Airflow
- Referencia de la API de Apache Airflow
- Operadores de Core Airflow en GitHub. Consulta la rama correspondiente a tu versión de Airflow.
- Operadores de Contrib Airflow en GitHub. Consulta la rama correspondiente a tu versión de Airflow.