Automatizar los pipelines: Airflow

Cursos Empresas Blog Live Conf Precios

Contenido del curso

Introducción a la ingeniería de datos

Herramientas del ciclo de DataOps

Ejerciendo como Data Engineer

Conclusiones

Tomar examen

Automatizar los pipelines: Airflow

Comentarios

Daviel Antonio Jiménez Comas

Estudiante

Mi resumen de la clase:

Para automatizar pipelines nos surgen tres preguntas: ¿Qué se va a correr? ¿En qué secuencia? ¿Cuándo y cada cuanto se va a correr?

Airflow nos ayuda a automatizar estas tareas, contestándonos las 3 preguntas. El que apoyándonos a conectar tareas de código que vamos a correr, conectándolas en secuencia y también con un manejador de tiempos para que se corra con una definición de cada cuánto se van a correr todo esto por medio de una interfaz web muy sencilla.

En el formato DAG es como le hablamos a Airflow para que ejecute las tareas en un grafo acíclico dirigido podemos indicarle cuales tareas irán al principio y cuales irán al final, poniendo los pasos con las dependencias adecuadas.

Ricardo Alanis

Profesor

Sergio Andrés Piratoba Forero

Estudiante

Great abstract!

Miguel Angel Reyes Moreno

Estudiante

Algo que noté y me pareció interesante es que hay código SQL dentro del código Python.

Ricardo Alanis

Profesor

Así es. Eso lo vas a encontrar! Hay prácticas donde se buscan separar y complementar con templates. Pero seguro lo encontrarás, especialmente en puntos que conectan con la base directamente (Como en la parte del Modelo en la arquitectura MVC)

Julian Castro Pulgarin

Estudiante

No conocía de Airflow ni del formato DAG, muy curioso, seguiré indagando. Yo he usado es Azure Data Factory para las pipelines, también es muy intuitivo y orientado a bloques

Ricardo Alanis

Profesor

Si! realmente lo importante es conectar bien la entrada, proceso y salida. El formato de dag me parece bastante práctico.

Antonio Demarco Bonino

Estudiante

Acá dejo el curso de Airflow para que lo vayan viendo: https://platzi.com/cursos/fundamentos-airflow/

Ricardo Alanis

Profesor

Eso, gracias por compartir mi Tony!

Nestor Alberto Malo Padilla

Estudiante

Le dije al jefe que iba a resolver el problema con un PipeLine, me dijo que: "No importa que sea Pipe o Lina que lo resuelva, quiero mi reporte!".

Ricardo Alanis

Profesor

Ian Cristian Ariel Yané

Estudiante

Wow, pense que seria con mil lineas de codigo mas

Ricardo Alanis

Profesor

Las suficientes <3

Cristian Ignacio Zuñiga Medina

Estudiante

Lo que mas rescato de la clase es el uso de 'Crons'. Algo nuevo para investigar que seguramente me será útil este semestre para sistemas operativos

Ricardo Alanis

Profesor

Correcto, programar tareas cada X tiempo y su monitoreo.

JEFERSON RIOS CALLE

Estudiante

Estoy siguiendo un ruta de Data Engineer ya establecida por platzi, sin embargo siento que esta mal estructurada ya que por ejemplo en este curso que es el primero de la ruta hablan ya de muchos conceptos que no conozco por lo que siento que este curso de Data Engineer o bien es para programadores experimentados o para aquellos que ya tiene una buena base de conocimientos y no para novatos que quieren aprender de este tema.

Me gustaría que Platzi colocase en sus rutas de cursos requisitos y/o conocimientos previos necesarios para iniciar cada curso y asi quizá no perder el tiempo

Ricardo Alanis

Profesor

Hola Jefferson! Gracias por tu comentario. Si tienes duda de algún concepto, no dudes en preguntar! Buscamos que cada contexto nuevo que mencionamos esté bien explicado, pero si te perdiste con alguno, aqui estamos para contestarte cualquier pregunta!

JOSE VICENTE DIAZ OSPINA

Estudiante

Muy interesante lo que se puede lograr con Airflow, espero llegar pronto al curso de la herramienta para poder entender mejor cada componente del Git que vimos en esta clase.

Ricardo Alanis

Profesor

Si, te va a gustar!

Excelena Denis Torin Gonzalez

Estudiante

Donde se encuentra el repositorio del proyecto de la clase? no veo ningún enlace

Cristian Camilo Quinche Vélez

Estudiante

Confieso que para mi ha sido difícil entender varios conceptos, este es mi primer acercamiento a la programación, he tratado de entender la lógica en el repositorio con los archivos que empiezan con s3, se ve que definen tareas en ciertos intervalos y cumplen una función, infiero que eso es un pipeline, desconozco si al final del proceso dejan la Data lista para ser interpretada.

Andres Correa

Estudiante

En mi trabajo utilizan Cron's y ahora se exactamente que son, estuve investigando por fuera que son y como crearlos y el termino es muy sencillo pero la cantidad de cosas que se me ocurrieron fue increíble, ahora estoy emocionadísimo de lo que se vendrá

Ricardo Alanis

Profesor

Si!! Vamos!

Camilo Granda Gómez

Estudiante

Airflow es una gran herramienta de orquestación de procesos. Es intuitiva y muy visual. Verán en el curso de Airflow lo bella que es.

Ricardo Alanis

Profesor

Asi es!

Luis Alberto Ramírez Castellanos

Estudiante

Interesante pero confuso.

Ricardo Alanis

Profesor

Cuentame más de lo que te pareció confuso!

Kevin Aguilar Contreras

Estudiante

•

¿Habrá un curso de Control M? para la automatización de un pipeline.

Laura Katherine Murcia Falla

Estudiante

Es mas o menos parecido a crear un job con SQL o un task scheduler con el asistente de windows? Gracias !!!

Ricardo Alanis

Profesor

Similar, pero con un manejo de tareas y logging mas visual y con una secuencia que suele tener entradas y salidas. Ese es el poder de los dags!

Yeimmy Andrea Ramirez Jimenez

Estudiante

Hola, dentro de la empresa en la que trabajo usan Control-M entendería que hace algo parecido a Airflow. Tienen alguna recomendación de donde puedo ampliar mis conocimientos de control M?

Ricardo Alanis

Profesor

¡Sí! Control-M y Airflow son dos herramientas similares que se usan en empresas para planificar y programar trabajos. Ambas tienen algunas diferencias y la elección depende de lo que necesite y prefiera cada empresa. Para ampliar tus conocimientos de Control-M, puedes hacer lo siguiente:

Revisa la documentación oficial de Control-M.
Busca recursos educativos en línea, como videos y foros.
Considera tomar cursos de capacitación.
Pide ayuda a colegas con experiencia en Control-M. Dentro de tu organización debe de existir experiencia en esto!

Kevin Acevedo

Estudiante

Ya empieza a coger sazoncita este tema de la ingeniería de datos.

Mario Alexander Vargas Celis

Estudiante

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.

Ruben Herrera

Estudiante

Acontinuacion presento de forma simple y estructurada mi análisis de Airflow: Apache Airflow es una herramienta para programar, monitorear y gestionar flujos de trabajo o pipelines de datos. En un entorno con repositorios, funciona de la siguiente manera:

Definición de workflows: Los flujos de trabajo se definen en código Python mediante archivos llamados DAGs (Directed Acyclic Graphs). Estos DAGs contienen tareas que se ejecutan en orden definido por dependencias.
Uso de repositorios:
- Los DAGs se almacenan en un repositorio, generalmente utilizando herramientas como Git.
- Esto facilita el versionado, colaboración entre equipos y despliegue en diferentes entornos (desarrollo, prueba, producción).
Programación y ejecución:
- Los DAGs son cargados en Airflow desde el repositorio (por ejemplo, mediante sincronización).
- Airflow programa las tareas en base a horarios o disparadores (triggers) definidos.
Ejecución distribuida:
- Las tareas pueden ejecutarse en diferentes nodos o máquinas, optimizando recursos y escalabilidad.
Monitoreo:
- Airflow incluye una interfaz web donde se pueden monitorear, reintentar o detener tareas, así como visualizar las dependencias entre ellas.

Ruben Herrera

Estudiante

Airflow actúa como un orquestador que toma los DAGs versionados, los programa y asegura que las tareas se ejecuten correctamente, proporcionando visibilidad y control sobre los flujos de trabajo.

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