Skip to main content

Ingeniería de Datos con Airflow

· 3 min read

Generator

La ingeniería de datos con Apache Airflow nos permite organizar una serie de tareas relacionadas con los datos. Con Airflow, puedes planificar y coordinar cada paso necesario para transformar y mover datos de un lugar a otro. Creas "diagramas" que indican qué tareas deben hacerse y en qué orden, como si fueran instrucciones de una receta. Cada tarea puede ser algo como recopilar datos de una fuente, transformarlos de cierta manera o almacenarlos en una base de datos.

Airflow se asegura de que las tareas se ejecuten en el orden correcto, se programen a la hora adecuada, y se puedan reintentar si algo sale mal. Airflow te ayuda a automatizar y supervisar todo el proceso de manejo de datos, haciéndolo más eficiente y confiable.

Tubería de datos ETL usando Apache Airflow​

Estos son unos simples pasos para crear una tubería con Airflow:

  1. Identifica las fuentes y destinos de los datos. ¿De qué fuentes de datos necesita extraer datos? ¿En qué destinos de datos necesita cargar los datos?
  2. Diseña un pipeline ETL. Debes determinar los pasos involucrados en la extracción, transformación y carga de los datos.
  3. Elije los operadores adecuados. Airflow proporciona una variedad de operadores para tareas ETL comunes, como extraer datos de bases de datos, cargar datos en almacenes de datos y transformar datos.
  4. Crea un DAG de flujo de aire. Los DAG (gráficos acíclicos dirigidos) de flujo de aire definen el flujo de trabajo de su canalización ETL.
  5. Configura el DAG. Establece la programación para el DAG y configure las dependencias entre tareas.
  6. Inicia el DAG. Una vez configurado el DAG, puede iniciarlo mediante la interfaz de usuario web o CLI de Airflow.

A continuación veamos un ejemplo simple de un DAG de Airflow para una canalización ETL que extrae datos de una base de datos PostgreSQL y los carga en una base de datos MySQL:

from airflow import DAG  
from airflow.operators.postgres_operator import PostgresOperator  
from airflow.operators.mysql_operator import MySqlOperator  
  
dag = DAG('etl_pipeline', start_date='2023-11-03')  
  
extract_task = PostgresOperator(  
task_id='extract_data',  
sql='SELECT * FROM customers',  
postgres_conn_id='postgres_default',  
dag=dag  
)  
  
load_task = MySqlOperator(  
task_id='load_data',  
sql='INSERT INTO customers (id, name, email) VALUES (%(id)s, %(name)s, %(email)s)',  
mysql_conn_id='mysql_default',  
params={'extract_task.output': extract_task.output},  
dag=dag  
)  
  
extract_task >> load_task

Este DAG extraerá datos de la customerstabla en la base de datos PostgreSQL y los cargará en la customerstabla en la base de datos MySQL.

También puede utilizar Airflow para automatizar canalizaciones ETL más complejas. Por ejemplo, puede utilizar Airflow para transformar datos antes de cargarlos en un almacén de datos o para cargar datos en múltiples destinos de datos.

Consejos adicionales para crear canalizaciones ETL con Airflow:​

  • Utilice variables y macros para hacer que sus DAG sean más dinámicos y reutilizables.
  • Utilice las funciones de activación y sensor de Airflow para iniciar DAG cuando ocurran ciertos eventos, como cuando se crea un nuevo archivo o cuando se inserta un nuevo registro en una base de datos.
  • Utilice las funciones de registro y métricas de Airflow para monitorear la ejecución de sus DAG y tareas.
  • Pruebe sus DAG con regularidad para asegurarse de que funcionen como se esperaba.

Si sigue estos consejos, podrá crear y mantener tuberías ETL eficientes y confiables con Airflow.