Pipeline con Apache Airflow + dbt: Caso Real con Código (2026)

Por qué Airflow + dbt se volvió el estándar

En 2026, la combinación Apache Airflow (orquestación) + dbt (transformación) es el stack dominante en pipelines analíticos modernos. Razones:

1. Airflow es madura (10+ años), open-source, con ecosistema masivo
2. dbt resolvió el problema de transformaciones mantenibles en SQL
3. Ambos tienen versiones cloud (MWAA, Astronomer, dbt Cloud)
4. Curva de aprendizaje razonable

Arquitectura del caso

Concretera con 15 plantas, 200K transacciones/mes.

Fuentes (SAP B1, despacho, GPS, fiscal)
        ↓
Fivetran (ingesta → raw schema)
        ↓
[ Airflow DAG: orquestación ]
        ↓
dbt (raw → staging → marts)
        ↓
BigQuery (data warehouse)
        ↓
Looker Studio (BI)

Estructura del proyecto dbt

concretera-dbt/
├── models/
│   ├── staging/
│   │   ├── stg_sap_facturas.sql
│   │   ├── stg_sap_clientes.sql
│   │   └── stg_despacho_pedidos.sql
│   ├── marts/
│   │   ├── core/
│   │   │   ├── dim_cliente.sql
│   │   │   ├── dim_producto.sql
│   │   │   └── fct_ventas.sql
│   │   └── finance/
│   │       ├── fct_margen_planta.sql
│   │       └── fct_cobranza.sql
├── tests/
├── macros/
├── seeds/
└── dbt_project.yml

DAG de Airflow (ejemplo simplificado)

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.providers.fivetran.operators.fivetran import FivetranOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'concretera_daily_pipeline',
    default_args=default_args,
    schedule_interval='0 6 * * *',
    start_date=datetime(2025, 1, 1),
    catchup=False,
) as dag:

    # 1. Trigger Fivetran syncs
    sync_sap = FivetranOperator(
        task_id='sync_sap',
        connector_id='fivetran_connector_sap',
    )

    sync_despacho = FivetranOperator(
        task_id='sync_despacho',
        connector_id='fivetran_connector_despacho',
    )

    # 2. Run dbt staging
    dbt_staging = BashOperator(
        task_id='dbt_staging',
        bash_command='cd /opt/dbt && dbt run --select staging',
    )

    # 3. Run dbt marts
    dbt_marts = BashOperator(
        task_id='dbt_marts',
        bash_command='cd /opt/dbt && dbt run --select marts',
    )

    # 4. Run dbt tests
    dbt_test = BashOperator(
        task_id='dbt_test',
        bash_command='cd /opt/dbt && dbt test',
    )

    # 5. Refresh dashboards
    refresh_looker = BashOperator(
        task_id='refresh_looker',
        bash_command='python /opt/scripts/refresh_looker.py',
    )

    [sync_sap, sync_despacho] >> dbt_staging >> dbt_marts >> dbt_test >> refresh_looker

Ejemplo de modelo dbt

stg_sap_facturas.sql:

{{ config(materialized='view') }}

with source as (
    select * from {{ source('sap_raw', 'facturas') }}
),

renamed as (
    select
        doc_num as factura_id,
        card_code as cliente_id,
        doc_date::date as fecha_factura,
        doc_total as total,
        currency as moneda,
        _fivetran_synced as ingested_at
    from source
    where doc_date >= '2023-01-01'
      and doc_status != 'Cancelled'
)

select * from renamed

fct_ventas.sql (mart):

{{ config(
    materialized='table',
    partition_by={'field': 'fecha_factura', 'data_type': 'date'}
) }}

with facturas as (
    select * from {{ ref('stg_sap_facturas') }}
),

clientes as (
    select * from {{ ref('dim_cliente') }}
),

productos as (
    select * from {{ ref('dim_producto') }}
),

final as (
    select
        f.factura_id,
        f.fecha_factura,
        c.nombre_cliente,
        c.segmento,
        p.categoria_producto,
        f.total,
        case
            when c.segmento = 'Grande' then 'Estratégico'
            when f.total > 500000 then 'Premium'
            else 'Regular'
        end as categoria_factura
    from facturas f
    left join clientes c on f.cliente_id = c.cliente_id
)

select * from final

Tests en dbt

schema.yml:

version: 2

models:
  - name: fct_ventas
    tests:
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns:
            - factura_id
    columns:
      - name: factura_id
        tests:
          - not_null
          - unique
      - name: total
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

CI/CD con GitHub Actions

name: dbt CI

on:
  pull_request:
    paths:
      - 'models/**'
      - 'tests/**'

jobs:
  dbt-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - name: Install dbt
        run: pip install dbt-bigquery
      - name: Run dbt build on dev
        env:
          DBT_PROFILES_DIR: .
          GCP_KEY: ${{ secrets.GCP_KEY }}
        run: |
          dbt deps
          dbt build --target dev

Lecciones aprendidas (producción)

1. Modelar primero, optimizar después

Pasamos 3 meses peleando con performance. Resuelto con partition + clustering + vistas materializadas.

2. dbt tests = salvavidas

Detectamos 3 errores de datos crítcos antes de que llegaran a dashboards directivos.

3. Monitoring es obligatorio

Airflow alertas vía Slack cuando un DAG falla o toma 2× el tiempo normal.

4. Documentación auto-generada

dbt docs generate + hosting → catálogo accesible para todo el equipo.

5. Separar dev / staging / prod

3 proyectos BigQuery, 3 perfiles dbt. Siempre.

Costos reales

Alternativa open-source: MWAA (AWS) + dbt Core + selfservice = ~$25K MXN/mes pero con más operación.

FAQ

¿Airflow vs Prefect vs Dagster? Airflow es más maduro y tiene mejor ecosistema. Prefect y Dagster son más modernos pero comunidad más pequeña.

¿dbt Core o dbt Cloud? Core si tienes equipo técnico. Cloud si quieres UI, scheduler y CI/CD built-in.

¿Funciona con Snowflake también? Sí, dbt soporta Snowflake, BigQuery, Redshift, Databricks, Postgres. Ajustas el adapter.

Conclusión

Airflow + dbt no es la única opción, pero es la más probada en 2026 para pipelines analíticos. El caso real muestra que con ~$45K MXN/mes y 3 meses de construcción, tienes pipeline productivo.

Teseo Data Lab diseña e implementa stacks Airflow + dbt para empresas mexicanas.