Fluyendo más y mejor con Airflow 2.0.0.

Apache Airflow es una herramienta open source que nos permite crear de forma programática, planificar y monitorizar workflows. En este artículo anterior ya hablábamos sobre las principales características de Airflow y sus orígenes.

Apache Airflow es, sin duda, uno de los proyectos de código abierto más vivos e interesantes que hemos visto en los últimos tiempos. Bajo el paraguas de la Apache Software Foundation el proyecto ha crecido y ha madurado hasta convertirse en el proyecto de la ASF escrito en Python con más líneas de código y uno de los más seguidos y forkeados en GitHub.

El 17 de diciembre de 2020 ha sido liberada la versión 2.0.0 de Apache Airflow que incluye multitud de novedades y mejoras. En este artículo vamos a analizar y comentar algunas de las mejoras más significativas.

Micro Paquetes

Airflow 2.0.0 ha sufrido una reorganización en la forma de gestionar e instalar los paquetes para integrarse con servicios y herramientas de diferentes proveedores. Esta es una de las características más potentes de Airflow, su gran capacidad de integración debido al gran número de componentes desarrollados por la comunidad.

Para ello Airflow ha huido de una arquitectura monolítica y se estructura en un módulo core y 61 módulos (en este momento) que ofrecen funcionalidad para la integración con los diferentes proveedores. Algunos ejemplos de estos nuevos módulos serían: AWS, Cassandra, Celery, Databricks, Docker, Google Cloud, Jenkins, Neo4J, Snowflake, Spark y muchos más. La lista completa actualizada se puede encontrar en esta sección de la documentación.

La forma de instalar Airflow vía pip con los paquetes adicionales que queramos sería esta

pip install apache-airflow[postgres,google]

En este caso, instalaríamos el paquete de Postgres y el Google Cloud.

Y si queremos añadir paquetes adicionales a una instalación previa de Airflow, lo haríamos también con pip de la siguiente manera:

pip install apache-airflow-providers-amazon

Esto instalaría el paquete para Amazon.

Mejoras en el planificador

Uno de los elementos centrales de Airflow es el planificador o scheduler, encargado de “repartir” el trabajo de los diferentes workflow. En algunos escenarios de uso intensivo este elemento se había convertido en un cuello de botella debido a no tener la capacidad de escalar ni de ofrecer un servicio en alta disponibilidad.

Esta era una de las demandas más importantes en la comunidad que se ha visto satisfecha con el rediseño e implementación de un nuevo scheduler en Arflow 2.0.0 que sí ofrece estas características:

• Escalabilidad horizontal: ahora es posible lanzar las réplicas que necesitemos de nuestro planificador y de ese modo aumentar el ancho de banda de la capacidad de planificación de nuestro sistema de acuerdo a nuestras necesidades.
• Baja latencia: independientemente del número de planificadores que despleguemos, se han introducido optimizaciones que hacen que estos sean mucho más rápidos a la hora de despachar las tareas. Algunos benchmarkings hablan de una mejora en el rendimiento de 17x en función del caso de uso.
• Alta disponibilidad: si tenemos un despliegue en HA (high availability) con 2 o más planificadores el tiempo de recuperación en caso de error de uno de ellos será cero. Esto permite también realizar tareas de mantenimiento sin pérdida del servicio. En esta sección de la documentación podemos encontrar más información sobre cómo desplegar Airflow en HA.

Mejoras en el desarrollo de DAGs

Otra de las mejoras más interesantes que se han introducido es la posibilidad de apoyarnos en decoradores de Python para definir los dags y los tasks. Esto hace más limpio y legible el código, en particular si usamos el PythonOperator, uno de los más usados sin duda. También nos simplifica mucho el uso de XCom, el mecanismo que nos ofrece Airflow para pasar mensajes entre tareas.

Veamos un ejemplo de cómo implementaríamos una pequeña ETL con los nuevos decoradores @dag y @task.

from airflow.decorators import dag, task
from airflow.utils.dates import days_ago

@dag(default_args={'owner': 'airflow'}, schedule_interval=None, start_date=days_ago(2))
def tutorial_taskflow_api_etl():
   @task
   def extract():
       return {"1001": 301.27, "1002": 433.21, "1003": 502.22}

   @task
   def transform(order_data_dict: dict) -> dict:
       total_order_value = 0

       for value in order_data_dict.values():
           total_order_value += value

       return {"total_order_value": total_order_value}

   @task()
   def load(total_order_value: float):

       print("Total order value is: %.2f" % total_order_value)

   order_data = extract()
   order_summary = transform(order_data)
   load(order_summary["total_order_value"])

tutorial_etl_dag = tutorial_taskflow_api_etl()

Mejoras en el API

Airflow contaba con una API experimental en versiones anteriores, pero desde 2.0.0 se ha introducido una versión estable del API que soporta toda la funcionalidad y nos ofrece un interfaz REST para manejar Airflow. La implementación API sigue la especificación de OpenAPI y podemos encontrar la documentación completa aquí.

Mejoras en el interfaz

Airflow 2.0.0 ha sufrido varios cambios y mejoras en la interfaz gráfica, sin duda un elemento muy importante y diferencial para Airflow. Hay varios cambios en diferentes partes de Airflow que mejoran la usabilidad y ofrecen un interfaz más moderno e intuitivo. Pero de todos ellos sin duda el más destacable y esperado por los usuarios de Airflow es la opción de “Auto-refresh” en la “Graph View” que nos permite visualizar de forma automática la evolución de la ejecución de un workflow sin necesidad de hacer nada.

Smart sensors

Los Sensors son una característica muy interesante de Airflow que nos permite definir tareas que esperen a que ocurra determinado evento, por ejemplo que se cree cierto fichero o se inserte cierta fila en una bases de datos. Son una herramienta muy potente, pero que podía llegar a sobrecargar mucho nuestro Airflow y mermar su rendimiento.

En 2.0.0 se ha introducido una mejora en los Sensors, llamada Smart Sensors, con lo que se pretende optimizar y agilizar el proceso de gestión de los Sensors, y de esta manera obtener un mejor rendimiento. Este nuevo concepto introduce un proceso centralizado que coordina y ejecuta el chequeo de los sensores de una forma más eficiente.

Otros cambios

Se han introducido otra serie de mejoras menores, algunas de ellas relacionadas con temas de seguridad, otras con la mejor organización de la configuración, otras tantas en Operators concretos y algunas relacionadas con la usabilidad como la inclusión de los TaskGroups como concepto para agrupar conjuntos de tareas y poder simplificar su repetición y su visualización en el grafo.

Conclusión

Airflow 2.0.0 es la evolución de una fantástica herramienta open source y un ejemplo de éxito. Muchas de las mejoras van orientadas a madurar el producto y prepararlo mejor para su uso intensivo en un entorno empresarial. No es de extrañar que proveedores cloud de primera línea como Google Cloud y AWS ya estén ofreciendo Airflow como servicio a través de sus productos Cloud Composer y MWAA respectivamente.

Airflow se está convirtiendo en un standard de facto para la orquestación de workflows y se vislumbra un futuro muy prometedor gracias a la gran comunidad que tiene detrás, el apoyo de compañías clave y su uso cada vez más extensivo.