Automatizar los pipelines: Airflow

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

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

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

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:

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

Componentes principales de Airflow

1. 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.
2. 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.
3. Scheduler:
   • Se encarga de programar y coordinar la ejecución de tareas según el horario y las dependencias definidas en el DAG.
4. 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.
5. Interfaz Web:
   • Proporciona una vista gráfica para monitorear, reintentar, o gestionar tareas y DAGs.
6. 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

1. Automatización Completa:
   • Programación de tareas en horarios definidos.
   • Dependencias claras entre tareas para garantizar orden y consistencia.
2. Flexibilidad:
   • Los DAGs se escriben en Python, lo que permite usar lógica compleja en los flujos de trabajo.
3. Escalabilidad:
   • Airflow se integra con herramientas como Celery y Kubernetes para distribuir tareas en clústeres grandes.
4. 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:

1. Extrae datos de una API.
2. Transforma los datos en un DataFrame de Pandas.
3. 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

1. Paralelismo por tareas:
   • Si las tareas son independientes, Airflow puede ejecutarlas simultáneamente.
   • Ejemplo: Extraer datos de múltiples APIs al mismo tiempo.
2. 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.

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:

1. 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.
2. 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).
3. 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.
4. Ejecución distribuida:
   • Las tareas pueden ejecutarse en diferentes nodos o máquinas, optimizando recursos y escalabilidad.
5. Monitoreo:
   • Airflow incluye una interfaz web donde se pueden monitorear, reintentar o detener tareas, así como visualizar las dependencias entre ellas.

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.

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.

En los diferentes proyectos podemos ver un ejemplo de plataforma en airflow, ver que en todas las carpetas y la informacion usa phyton el cual es nuestro caballo de batalla y tambien podemos notar los diferentes comandos, podemos usar los codigos y ponerlos en la web de cron expresion para saber a que se refiere si no tenemos la experienci aun

Aquí hay una lista de empresas que usan Spark y Spark Stream

• Uber
• Netflix
• Airbnb
• Alibaba
• eBay
• Amazon
• Yahoo
• IBM
• Salesforce
• TripAdvisor
• Spotify
• Pinterest
• Zillow
• Yelp
• NASA
• LinkedIn
• Twitter
• Tencent
• Capital One
• Cisco

Esta es la primera vez que escucho de estas herramientas, leyendo algunos blogs me doy cuenta que el poder de las mismas es increíble, cada vez más me está atrayendo Data Engineering

Airflow es un símil de Azure Data Factory/Bricks??

Pipeline: Extraer informacion, transformarla y dejarla en el lugar adecuado

Gracias

Estuve usando talend y aunque es muy buena herramienta, Airflow me está gustando bastante.

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á

Si!! Vamos!

✅

<3

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!".

xd

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

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!

12. Automatizar los pipelines: Airflow

Pipeline: ETL

Tres preguntas importantes

• ¿Qué va a correr?
• ¿En qué secuencia?
• ¿Cuándo y cada cuánto va a correr?

Airflow

• Responde las 3
• DAG, como le hablamos a airflow

<3

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?

¡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!

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.

Si, te va a gustar!