Instalar Airflow con Docker: el estándar de la industria

### ¿Por qué Airflow si ya tenemos Step Functions?

Empezamos con Step Functions porque es conceptualmente más simple: máquinas de estado, diagramas de flujo, JSON declarativo. Pero ahora necesitas conocer el estándar de la industria. Si buscas "data engineer" en LinkedIn, el 70% de las ofertas mencionan Airflow. No porque sea mejor que Step Functions — son herramientas diferentes para contextos diferentes — sino porque Airflow lleva más tiempo, es open source, funciona con cualquier cloud y tiene un ecosistema enorme.

Airflow nació en Airbnb en 2014. Los ingenieros de datos tenían docenas de pipelines en cron, se perdían dependencias, no había visibilidad de qué había corrido y qué no. Necesitaban algo que definiera dependencias entre tareas, las ejecutara en orden, registrara el historial y tuviera una UI para monitorizar. Así nació Airflow — y desde entonces se ha convertido en el estándar de facto.

La diferencia filosófica: Step Functions piensa en máquinas de estado (diagrama de flujo). Airflow piensa en DAGs — grafos dirigidos acíclicos. Son como un árbol genealógico: cada tarea sabe quién son sus padres (dependencias) y no puede haber ciclos (A depende de B que depende de A). El scheduler mira el grafo y ejecuta cada tarea cuando sus dependencias están satisfechas.

### Paso 1: Crear la carpeta y el docker-compose.yml

Vamos a usar la versión "standalone" que es más simple para aprender (un solo contenedor en lugar de los 9 que tiene el compose oficial).

1# Crear carpeta para Airflow
2mkdir -p airflow-lab/dags airflow-lab/logs airflow-lab/plugins
3cd airflow-lab

1# docker-compose.yml para Airflow standalone
2services:
3  airflow:
4    image: apache/airflow:2.8.1-python3.11
5    container_name: airflow-standalone
6    environment:
7      - AIRFLOW__CORE__EXECUTOR=SequentialExecutor
8      - AIRFLOW__CORE__LOAD_EXAMPLES=False
9      - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=True
10      - _AIRFLOW_DB_MIGRATE=true
11      - _AIRFLOW_WWW_USER_CREATE=true
12      - _AIRFLOW_WWW_USER_USERNAME=admin
13      - _AIRFLOW_WWW_USER_PASSWORD=admin
14    volumes:
15      - ./dags:/opt/airflow/dags
16      - ./logs:/opt/airflow/logs
17      - ./plugins:/opt/airflow/plugins
18    ports:
19      - "8080:8080"
20    command: airflow standalone
21    healthcheck:
22      test: ["CMD", "curl", "--fail", "http://localhost:8080/health"]
23      interval: 30s
24      timeout: 10s
25      retries: 5

Este docker compose hace varias cosas: crea un usuario admin/admin para la UI web, desactiva los DAGs de ejemplo (que son confusos para principiantes), usa SequentialExecutor (un solo worker — suficiente para aprender), y monta tus carpetas locales como volúmenes.

### Paso 2: Levantar Airflow

1# Levantar Airflow (la primera vez tarda 1-2 minutos en inicializar la DB)
2docker compose up -d
3
4# Ver los logs mientras arranca
5docker compose logs -f airflow
6
7# Esperar hasta ver algo como:
8# "Airflow is ready"
9# "webserver | [INFO] Listening at: http://0.0.0.0:8080"
10
11# Verificar que está healthy
12docker ps
13# STATUS: Up X minutes (healthy)

### Paso 3: Acceder a la interfaz web

Abre tu navegador en http://localhost:8080. Verás el login de Airflow. Credenciales: admin / admin. Una vez dentro, verás el dashboard de DAGs (vacío porque aún no hemos creado ninguno).

Si admin/admin no te deja entrar, es que el modo standalone genero una contrasena aleatoria en vez de usar la del compose. La encuentras asi: docker exec airflow-standalone cat /opt/airflow/standalone_admin_password.txt

La UI de Airflow es una de sus grandes ventajas sobre Step Functions: puedes ver el historial de ejecuciones de cada DAG, los logs de cada tarea, el estado actual, y lanzar/pausar DAGs con un click. Es como un centro de control para todos tus pipelines.

### Paso 4: Tu primer DAG — Hello World

Un DAG en Airflow es un archivo Python que vive en la carpeta dags/. Airflow escanea esa carpeta cada 30 segundos buscando archivos que definan DAGs. Vamos a crear el más simple posible:

1# dags/hello_world.py
2from datetime import datetime
3from airflow import DAG
4from airflow.operators.python import PythonOperator
5
6def saludar():
7    """Función que ejecutará nuestra tarea."""
8    print("¡Hola desde Airflow! El orquestador está funcionando.")
9    return "Saludo completado"
10
11# Definir el DAG
12with DAG(
13    dag_id='hello_world',
14    description='Mi primer DAG — verifica que Airflow funciona',
15    start_date=datetime(2024, 1, 1),
16    schedule=None,  # Solo se ejecuta manualmente (por ahora)
17    catchup=False,
18    tags=['tutorial'],
19) as dag:
20    
21    tarea_saludo = PythonOperator(
22        task_id='saludar',
23        python_callable=saludar,
24    )

Desglosemos el código: with DAG(...) define el contenedor del pipeline. Dentro creamos tareas usando Operators (en este caso PythonOperator, que ejecuta una función Python). El schedule=None significa que no se ejecuta automáticamente — lo lanzamos a mano desde la UI. catchup=False evita que ejecute todas las fechas pasadas desde start_date.

### Paso 5: Ejecutar y verificar

1# Verificar que Airflow detectó el DAG
2docker exec airflow-standalone airflow dags list
3
4# Deberías ver:
5# dag_id      | filepath         | owner   | paused
6# hello_world | hello_world.py   | airflow | True
7
8# Activar el DAG (despausa)
9docker exec airflow-standalone airflow dags unpause hello_world
10
11# Ejecutar manualmente
12docker exec airflow-standalone airflow dags trigger hello_world
13
14# Ver el estado de la última ejecución
15docker exec airflow-standalone airflow dags list-runs -d hello_world

Si vas a la UI web y haces click en hello_world → Graph, verás tu DAG como un grafo con un solo nodo verde (éxito). Haz click en el nodo → Log para ver el print de tu función. Si ves "¡Hola desde Airflow!", tu instalación está completa.

### Estructura de Airflow: qué componentes están corriendo

• ▹Scheduler: el cerebro. Escanea los DAGs, decide qué tareas deben ejecutarse según el schedule y las dependencias.
• ▹Webserver: la UI web en el puerto 8080. Solo monitorización y control manual.
• ▹Metadata DB (SQLite en standalone): almacena historial de ejecuciones, estado de tareas, configuración.
• ▹Executor: el motor que ejecuta las tareas. SequentialExecutor las ejecuta una a una (suficiente para aprender).

La version esta clavada en apache/airflow:2.8.1-python3.11 a proposito, y no es solo higiene. La version estable de hoy es Airflow 3, y tres piezas de esta leccion no existen alli: PythonOperator se importa de otro paquete, el SequentialExecutor se ha eliminado, y el webserver se llama api-server. Todo lo que aprendes aqui vale igual — DAGs, operadores, dependencias — pero si pones :latest te vas a encontrar imports que no compilan. Cuando quieras dar el salto, busca la guia "Upgrading to Airflow 3".

Una cosa que ya estas haciendo sin saberlo: cuando tu funcion devuelve un valor (como {"registros": 1000}), Airflow lo guarda en XCom, su buzon interno para pasar valores pequenos entre tareas. La siguiente tarea lo puede leer con ti.xcom_pull(task_ids="extraer"). Ojo: XCom va a la base de datos de metadatos, asi que ahi viajan contadores, rutas de S3 e ids — nunca los datos. Es la misma regla que en Step Functions. Lo veras en detalle en la siguiente leccion.

### Comandos útiles del día a día

1# Listar DAGs disponibles
2docker exec airflow-standalone airflow dags list
3
4# Ejecutar un DAG manualmente
5docker exec airflow-standalone airflow dags trigger <dag_id>
6
7# Ver estado de tareas de una ejecución
8docker exec airflow-standalone airflow tasks list <dag_id>
9
10# Pausar/despausar un DAG
11docker exec airflow-standalone airflow dags pause <dag_id>
12docker exec airflow-standalone airflow dags unpause <dag_id>
13
14# Ver logs de una tarea específica
15docker exec airflow-standalone airflow tasks test <dag_id> <task_id> <logical_date>
16# Ejecuta la tarea al momento e imprime su salida (no crea DagRun)
17
18# Los logs de ejecuciones pasadas estan en la carpeta que montaste:
19# ls logs/dag_id=<dag_id>/
20
21# Parar Airflow
22docker compose down
23
24# Parar y borrar todo (incluida la DB)
25docker compose down -v