15  Orquestación y Herramientas Modernas

Una vez dominado PySpark, el siguiente paso es automatizar los pipelines y adoptar las herramientas que usa la industria para transformaciones declarativas y scheduling de producción.

15.1 ETL moderno: dbt + PySpark

dbt (data build tool) separa la transformación del pipeline técnico: defines las transformaciones como modelos SQL, y dbt se encarga de ejecutarlos en el orden correcto, con pruebas automáticas y documentación.

pip install dbt-spark
dbt init mi_proyecto_etl
-- models/staging/stg_viajes.sql
-- dbt materializa esto como una tabla o vista en Spark/DW
SELECT
    VendorID                    AS vendor_id,
    tpep_pickup_datetime        AS pickup_at,
    trip_distance,
    fare_amount,
    tip_amount,
    total_amount,
    payment_type
FROM {{ source('raw', 'yellow_taxi') }}
WHERE trip_distance > 0
  AND fare_amount   > 2.5
-- models/marts/fct_viajes_diarios.sql
SELECT
    DATE(pickup_at)             AS fecha,
    HOUR(pickup_at)             AS hora,
    COUNT(*)                    AS n_viajes,
    AVG(total_amount)           AS ticket_prom,
    SUM(total_amount)           AS ingreso_total
FROM {{ ref('stg_viajes') }}
GROUP BY 1, 2
# models/schema.yml — pruebas automáticas de calidad
models:
  - name: stg_viajes
    columns:
      - name: vendor_id
        tests: [not_null]
      - name: total_amount
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 500
dbt run      # ejecuta los modelos
dbt test     # corre las pruebas de calidad
dbt docs generate && dbt docs serve   # abre documentación interactiva

15.2 Orquestación con Apache Airflow

En producción, los pipelines ETL se ejecutan automáticamente según un calendario. Airflow define estos flujos como DAGs de Python:

# dags/etl_taxi_mensual.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def extraer(**ctx):
    mes = ctx['logical_date'].strftime("%Y-%m")
    print(f"Extrayendo mes {mes}...")
    # aquí va el código de extracción

def transformar(**ctx): ...
def cargar(**ctx): ...
def validar(**ctx): ...

with DAG(
    dag_id="etl_nyc_taxi_mensual",
    start_date=datetime(2023, 1, 1),
    schedule_interval="0 3 1 * *",  # el día 1 de cada mes a las 3 AM
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
) as dag:

    t_extract   = PythonOperator(task_id="extraer",     python_callable=extraer)
    t_transform = PythonOperator(task_id="transformar", python_callable=transformar)
    t_load      = PythonOperator(task_id="cargar",      python_callable=cargar)
    t_validate  = PythonOperator(task_id="validar",     python_callable=validar)

    t_extract >> t_transform >> t_load >> t_validate
Tip

Alternativas a Airflow: - Prefect — más Python-nativo, sin XML/YAML, fácil para equipos pequeños. - Dagster — fuerte en observabilidad y gestión de assets de datos. - Para empezar: Prefect + PySpark local → luego migrar a Airflow en producción.

15.3 Resumen y mapa de herramientas

Haz clic en una herramienta para ver cuándo usarla.

Etapa Herramienta Cuándo usarla
Extract PySpark Archivos grandes (>1 GB), clústeres, Parquet/JDBC
Extract Kafka + Debezium Streaming en tiempo real, CDC
Transform Spark SQL / DataFrame API Joins complejos, window functions, ML features
Transform dbt Transformaciones SQL versionadas sobre DW cloud
Quality Great Expectations Validación declarativa antes de la carga
Load Delta Lake / Iceberg ACID, upsert, time travel en el lakehouse
Load BigQuery / Snowflake DW cloud sin gestión de infraestructura
Load DuckDB Notebooks, sin clúster, prototipos rápidos
Orquestar Airflow Pipelines de producción complejos con SLA
Orquestar Prefect Equipos pequeños, Python-native, simplicidad

15.4 Referencias y recursos adicionales