Automatizar los pipelines: Airflow

Curso de Fundamentos de Ingeniería de Datos

Contenido del curso

Introducción a la ingeniería de datos

Herramientas del ciclo de DataOps

Ejerciendo como Data Engineer

Conclusiones

Tomar examen

Cursos Empresas Blog Live Conf Eventos Precios

0:005:00

Automatizar los pipelines: Airflow

Curso de Fundamentos de Ingeniería de Datos

Clase 12 de 25

Clase 12 / 25

Mario Alexander Vargas Celis

Hace 2 años

Automatización de Pipelines con Apache Airflow

Apache Airflow es una herramienta de orquestación de flujos de trabajo que permite automatizar la ejecución, monitoreo y mantenimiento de pipelines de datos. Es ampliamente utilizado en el mundo de la ingeniería de datos para gestionar tareas de transformación, carga, y análisis de grandes volúmenes de información de manera eficiente.

¿Qué es Apache Airflow?

Airflow es una plataforma de código abierto diseñada para:

Crear y programar flujos de trabajo complejos (pipelines) de manera declarativa utilizando Python.
Monitorear y gestionar el estado de los flujos mediante una interfaz gráfica web.
Escalar pipelines a entornos de producción distribuidos.

Componentes principales de Airflow

DAG (Directed Acyclic Graph):
- Es la estructura principal de un pipeline en Airflow.
- Representa tareas como nodos y las dependencias entre ellas como aristas.
- Los DAGs deben ser acíclicos (sin ciclos) para asegurar que las tareas se ejecuten en el orden correcto.
Tasks (Operadores):
- Cada tarea es una unidad de trabajo definida en un DAG.
- Los operadores son funciones predefinidas para ejecutar acciones específicas:
  - BashOperator: Ejecutar comandos de shell.
  - PythonOperator: Ejecutar funciones de Python.
  - PostgresOperator: Ejecutar consultas SQL en bases de datos PostgreSQL.
  - S3Operator: Interactuar con Amazon S3.
Scheduler:
- Se encarga de programar y coordinar la ejecución de tareas según el horario y las dependencias definidas en el DAG.
Executor:
- Gestiona cómo y dónde se ejecutan las tareas. Ejemplos:
  - LocalExecutor: Ejecuta tareas en el mismo nodo.
  - CeleryExecutor: Escala tareas en múltiples nodos.
Interfaz Web:
- Proporciona una vista gráfica para monitorear, reintentar, o gestionar tareas y DAGs.
Metadatos y Base de Datos:
- Airflow utiliza una base de datos para almacenar información sobre el estado de las tareas y DAGs.

Ventajas de Airflow

Automatización Completa:
- Programación de tareas en horarios definidos.
- Dependencias claras entre tareas para garantizar orden y consistencia.
Flexibilidad:
- Los DAGs se escriben en Python, lo que permite usar lógica compleja en los flujos de trabajo.
Escalabilidad:
- Airflow se integra con herramientas como Celery y Kubernetes para distribuir tareas en clústeres grandes.
Integración con Ecosistemas de Big Data:
- Compatible con bases de datos (SQL/NoSQL), herramientas de cloud (AWS, GCP, Azure) y frameworks de big data como Spark, Hadoop, o Kafka.

Ejemplo de un Pipeline en Airflow

Supongamos que queremos automatizar un pipeline ETL que:

Extrae datos de una API.
Transforma los datos en un DataFrame de Pandas.
Carga los datos procesados a una base de datos PostgreSQL.

Código del DAG en Python:

from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime import requests import pandas as pd import psycopg2

# Función para extraer datos def extract_data(): response = requests.get("https://api.example.com/data") data = response.json() pd.DataFrame(data).to_csv("/tmp/raw_data.csv", index=False)

# Función para transformar datos def transform_data(): df = pd.read_csv("/tmp/raw_data.csv") df["new_column"] = df["old_column"].apply(lambda x: x * 2) df.to_csv("/tmp/transformed_data.csv", index=False)

# Función para cargar datos a PostgreSQL def load_data(): df = pd.read_csv("/tmp/transformed_data.csv") conn = psycopg2.connect( host="localhost", database="example_db", user="username", password="password" ) cursor = conn.cursor() for _, row in df.iterrows(): cursor.execute( "INSERT INTO processed_data (column1, column2) VALUES (%s, %s)", (row["column1"], row["new_column"]) ) conn.commit() cursor.close() conn.close()

# Definición del DAG default_args = { "owner": "data_engineer", "retries": 3, "retry_delay": timedelta(minutes=5), } with DAG( dag_id="etl_pipeline", default_args=default_args, start_date=datetime(2025, 1, 1), schedule_interval="0 12 * * *", # Ejecutar diariamente a las 12 PM catchup=False, ) as dag:

extract_task = PythonOperator( task_id="extract_data", python_callable=extract_data, )

transform_task = PythonOperator( task_id="transform_data", python_callable=transform_data, )

load_task = PythonOperator( task_id="load_data", python_callable=load_data, )

# Definimos las dependencias extract_task >> transform_task >> load_task

Cómputo paralelo con Airflow

Paralelismo por tareas:
- Si las tareas son independientes, Airflow puede ejecutarlas simultáneamente.
- Ejemplo: Extraer datos de múltiples APIs al mismo tiempo.
Executor avanzado:
- Con CeleryExecutor o KubernetesExecutor, Airflow escala tareas en clústeres distribuidos.

Empresas que usan Airflow

Airbnb (creador original de Airflow).
Uber: Automatización de pipelines de datos en tiempo real.
Netflix: Procesamiento de datos para recomendaciones.
Slack: Análisis de datos internos.
Lyft: Orquestación de pipelines de datos de movilidad.

Conclusión

Apache Airflow es una herramienta esencial para la automatización de pipelines de datos. Su flexibilidad, escalabilidad y capacidad de integración con herramientas modernas lo convierten en una opción líder para tareas ETL, procesamiento distribuido y orquestación de flujos de trabajo complejos en proyectos de big data.

Comentarios (30)30 comentarios

Daviel Antonio Jiménez Comas

Hace 4 años

Ricardo Alanis

Profesor

Sergio Andrés Piratoba Forero

Miguel Angel Reyes Moreno

Julian Castro Pulgarin

Antonio Demarco Bonino

Nestor Alberto Malo Padilla

Hace 3 años

Ian Cristian Ariel Yané

Cristian Ignacio Zuñiga Medina

Excelena Denis Torin Gonzalez

Hace un año

JEFERSON RIOS CALLE

JOSE VICENTE DIAZ OSPINA

Kevin Aguilar Contreras

Hace un año•editado

Cristian Camilo Quinche Vélez

Andres Correa

Camilo Granda Gómez

Luis Alberto Ramírez Castellanos

Laura Katherine Murcia Falla

Yeimmy Andrea Ramirez Jimenez

Kevin Acevedo

Ruben Herrera

Introducción a la ingeniería de datos

¿Qué es ingeniería de datos? ¿Qué es Data Engineer?

Formación Integral para Data Engineers: Retos y Herramientas Esenciales

¿Cómo convertirte en Data Engineer?

¿Dónde ejercer como Data Engineer?

Tareas de Data Engineer: DataOPs

Herramientas del ciclo de DataOps

Agile en ingeniería de datos

Lenguajes de programación e ingeniería de software

¿Dónde y cómo escribir tu código en ingeniería de datos?

Automatización y scripting

Fuentes de datos: SQL, NoSQL, API y web scraping

Procesamiento de datos: pipelines, Apache Spark y cómputo paralelo

Automatizar los pipelines: Airflow

Containers y empaquetamiento: Docker y Kubernetes

Manejo de ambientes para datos

Testing de software y de datos

CI/CD basico

Servidores y computación en la nube para data

Reentrenamiento y control de salud de servicios

Medición de indicadores y seguimiento a proyectos

Ejerciendo como Data Engineer

Buscando Oportunidades como Data Engineer

Evolución en el rol: ganando seniority como Data Engineer

Evolución en el rol: manager, architect, pivot

Trabajando en equipo como Data Engineer

Conclusiones

Compartir con la comunidad y seguir creciendo

Completa tu guía con los retos y certifícate

Automatizar los pipelines: Airflow

Progreso del curso

Introducción a la ingeniería de datos

Herramientas del ciclo de DataOps

Ejerciendo como Data Engineer

Conclusiones

Recursos

Recursos