Transformar datos en el warehouse con dbt

Resumen

dbt (data build tool) es un framework de transformación SQL-first que convierte tu data warehouse en una plataforma de analytics engineering versionada. Escribís sentencias SELECT en archivos .sql y dbt se ocupa de la materialización (table, view, incremental), la resolución de dependencias, los tests y la documentación. dbt compila tus modelos a SQL, los ejecuta en el warehouse (BigQuery, Snowflake, Redshift, Postgres) y registra el linaje entre modelos.

Cuándo usarlo

• Transformar datos crudos del warehouse en tablas listas para analytics
• Cuando querés transformaciones SQL versionadas y testeables
• Para armar una arquitectura de modelos por capas (staging → intermediate → marts)
• Equipos donde los analistas escriben SQL pero necesitan prácticas de ingeniería de software
• Cuando necesitás linaje de datos, documentación y freshness checks

Cuándo no usarlo

• Transformaciones en tiempo real/streaming — dbt es batch
• Datasets chicos en pandas — usá pandas/Polars directamente
• ETL donde la extracción y la carga son el cuello de botella — dbt solo hace la T de ETL
• Cuando necesitás lógica procedural compleja (loops, condicionales) — usá stored procedures o Python. Si extracción y carga dominan y la T es trivial, un pipeline ETL con pandas es una opción más liviana.

Explicación

Cómo corre un dbt run

dbt run sigue una secuencia fija:

1. Parsea y compila — dbt lee cada .sql, resuelve el Jinja ({{ ref() }}, {{ source() }}, {{ config() }}) y genera el SQL crudo que ejecuta tu warehouse.
2. Construye el DAG — los nodos source() son tablas externas, las aristas ref() son dependencias entre modelos. dbt ordena la ejecución desde ese DAG: un modelo nunca corre antes de que terminen sus padres.
3. Materializa — cada nodo corre su SQL compilado como CREATE VIEW, CREATE TABLE o MERGE según su materialización, en el schema del target.

Cada ref() es una arista de ese grafo — por eso los modelos referencian nombres de modelo y no nombres de tabla. Si stg_customers falla, fct_orders y dim_customers se saltan en vez de construirse sobre datos rotos, y dbt run reporta todo el subárbol como skipped.

Elegir la materialización

Materialización	Crea	Coste por corrida	Usala para
view	Una vista	~0 (no escribe datos)	Staging; modelos consumidos a través de otros
table	Una tabla física	Rebuild completo cada corrida	Marts consultados directo y seguido
incremental	Tabla + merge/append de filas nuevas	Solo datos nuevos o cambiados	Facts grandes con unique_key confiable
ephemeral	Nada — se inyecta como CTE downstream	Solo coste de compilación	Pasos intermedios usados por uno o dos modelos

La disposición de esta receta — staging como views, intermediate como ephemeral, marts como tablas con el fact grande incremental — asigna a cada capa la materialización más barata que cumple su función.

Qué corre dbt test realmente

Los tests de esquema compilan a aserciones SQL: unique se convierte en GROUP BY ... HAVING COUNT(*) > 1, not_null en un WHERE col IS NULL, relationships en un anti-join contra el modelo referenciado. Un test que devuelve filas, falla. Desde dbt 1.2, dbt build corre modelos y sus tests juntos en orden de DAG — un modelo que falla sus tests bloquea a sus hijos downstream, que es exactamente el comportamiento que querés en CI. Para validaciones más allá del esquema (distribuciones, rangos, reglas de negocio), un validador del lado Python como la receta de Pandera complementa los tests de dbt.

Solución

Setup del proyecto

pip install dbt-core dbt-postgres   # adaptador según el warehouse: dbt-snowflake, dbt-bigquery...
dbt init my_project                  # crea dbt_project.yml y el esqueleto de carpetas

profiles.yml (en ~/.dbt/ por defecto) guarda la conexión — son credenciales, así que queda fuera del versionado:

# ~/.dbt/profiles.yml
my_project:
  target: dev            # entorno por defecto
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: analytics
      pass: "{{ env_var('DB_PASSWORD') }}"
      dbname: warehouse
      schema: dbt_dev
      threads: 4

Estructura de proyecto

dbt_project/
├── dbt_project.yml
├── profiles.yml
├── models/
│   ├── staging/
│   │   ├── stg_orders.sql
│   │   ├── stg_customers.sql
│   │   └── schema.yml
│   ├── intermediate/
│   │   └── int_orders_enriched.sql
│   └── marts/
│       ├── fct_orders.sql
│       ├── dim_customers.sql
│       └── schema.yml
├── macros/
│   └── cents_to_dollars.sql
├── tests/
│   └── assert_order_status.sql
└── snapshots/
    └── snap_customers.sql

Modelo básico

-- models/staging/stg_orders.sql
SELECT
    order_id::integer AS order_id,
    customer_id::integer AS customer_id,
    order_date::date AS order_date,
    amount::numeric(10,2) AS amount,
    status::varchar AS status
FROM {{ source('raw', 'orders') }}

Definición de sources

# models/staging/schema.yml
version: 2

sources:
  - name: raw
    database: warehouse
    schema: raw_data
    freshness:
      warn_after: {count: 12, period: hour}
      error_after: {count: 24, period: hour}
    loaded_at_field: _loaded_at
    tables:
      - name: orders
        columns:
          - name: order_id
            tests:
              - unique
              - not_null
          - name: amount
            tests:
              - not_null
      - name: customers
        columns:
          - name: customer_id
            tests:
              - unique
              - not_null

Modelo con referencias

-- models/marts/fct_orders.sql
SELECT
    o.order_id,
    o.customer_id,
    o.order_date,
    o.amount,
    o.status,
    c.customer_name,
    c.customer_tier,
    c.city
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref('stg_customers') }} c
    ON o.customer_id = c.customer_id
WHERE o.status = 'completed'

ref() crea una dependencia — dbt corre stg_orders y stg_customers antes de fct_orders. El bloque freshness del source, junto a loaded_at_field, hace que dbt source freshness avise cuando la carga upstream llega tarde.

Materializaciones

# dbt_project.yml
models:
  my_project:
    staging:
      +materialized: view
    intermediate:
      +materialized: ephemeral
    marts:
      +materialized: table
      fct_orders:
        +materialized: incremental

-- models/marts/fct_orders.sql — modelo incremental
{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge'
) }}

SELECT
    o.order_id,
    o.customer_id,
    o.order_date,
    o.amount
FROM {{ ref('stg_orders') }} o
WHERE o.status = 'completed'

{% if is_incremental() %}
    AND o.order_date > (SELECT MAX(order_date) FROM {{ this }})
{% endif %}

Tests de esquema

# models/marts/schema.yml
version: 2

models:
  - name: fct_orders
    description: "Órdenes completadas con datos de cliente"
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: amount
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: "amount >= 0"
      - name: status
        tests:
          - accepted_values:
              values: ['completed', 'pending', 'cancelled']

Test singular custom

-- tests/assert_no_negative_amounts.sql
SELECT *
FROM {{ ref('fct_orders') }}
WHERE amount < 0

Macros

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name) %}
    ({{ column_name }} / 100.0)
{% endmacro %}

-- Uso del macro en un modelo
SELECT
    order_id,
    {{ cents_to_dollars('amount_cents') }} AS amount_dollars
FROM {{ ref('stg_orders') }}

Snapshots para SCD tipo 2

-- snapshots/snap_customers.sql
{% snapshot snap_customers %}
{{
    config(
      target_schema='snapshots',
      unique_key='customer_id',
      strategy='timestamp',
      updated_at='updated_at',
    )
}}
SELECT * FROM {{ source('raw', 'customers') }}
{% endsnapshot %}

Los snapshots registran los cambios de los datos del source a lo largo del tiempo, creando una tabla de historial con las columnas valid_from, valid_to y dbt_scd_id.

Uso de variables

-- models/marts/fct_orders.sql
SELECT *
FROM {{ ref('stg_orders') }}
WHERE order_date >= '{{ var('start_date', '2025-01-01') }}'

dbt run --vars '{"start_date": "2025-06-01"}'

Hooks

# dbt_project.yml
on-run-start:
  - "CREATE SCHEMA IF NOT EXISTS {{ target.schema }}"
on-run-end:
  - "{{ log('Model run complete', info=True) }}"

Variantes

El proyecto dbt completo — todos los modelos, tests, el snapshot, el macro, la plantilla de profiles.yml y el DAG de Airflow — está en el repositorio complementario.

dbt con Airflow

Una sola tarea dbt build cubre run + tests en orden de DAG (dbt ≥1.2). Si necesitás tareas separadas de run/test para alertas, mantené dos BashOperators:

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

dag = DAG("dbt_pipeline", schedule="@daily", start_date=datetime(2025, 1, 1))

dbt_build = BashOperator(
    task_id="dbt_build",
    bash_command="cd /opt/dbt && dbt build --select marts.*",
    dag=dag,
)

Para una integración más rica — una tarea por modelo, reintentos por modelo, parseo parcial — mirá astronomer-cosmos, el provider de Airflow que renderiza el DAG de dbt dentro de Airflow.

Materialización custom con post-hook

{{ config(
    materialized='table',
    post_hook="CREATE INDEX IF NOT EXISTS idx_orders_date ON {{ this }} (order_date)"
) }}

SELECT * FROM {{ ref('stg_orders') }}

Paquetes

# packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: 1.1.1
  - package: calogica/dbt_expectations
    version: 0.10.3

dbt deps

-- Uso de dbt_utils
{{ dbt_utils.date_spine(
    datepart="day",
    start_date="'2025-01-01'",
    end_date="'2025-12-31'"
) }}

Buenas prácticas

• Usá arquitectura en capas: staging (limpieza) → intermediate (joins) → marts (lógica de negocio)
• Testeá siempre las claves primarias con unique y not_null — atrapa problemas de calidad temprano
• Usá materialización ephemeral para modelos intermedios usados una sola vez — no ocupa storage
• Usá incremental para tablas de hechos grandes — procesa solo filas nuevas, no todo el historial
• Documentá cada modelo y columna — dbt docs generate crea un sitio de documentación
• Usá ref() en vez de hardcodear nombres de tabla — dbt resuelve dependencias y orden
• Usá source() para datos crudos y ref() para modelos dbt — separa lo externo de lo interno
• Corré dbt test (o dbt build) en CI — que el pipeline falle ante problemas de calidad
• Cuando el pipeline crece más allá de dbt run, scheduléalo y monitorealo con Airflow

Errores comunes

• No testear los modelos: sin tests, los datos malos se propagan en silencio. Testeá siempre claves primarias y campos críticos.
• Usar table para todo: las vistas son más baratas para modelos consultados poco. Usá table solo para modelos consultados seguido.
• No usar incremental en tablas grandes: reconstruir 100M de filas en cada corrida es lento. Usá incremental con unique_key.
• Hardcodear nombres de schema: usá {{ target.schema }} o {{ this }} — así desplegás a dev/staging/prod sin tocar los modelos.
• No usar paquetes: dbt_utils y dbt_expectations traen macros testeadas para patrones comunes.

Lectura adicional

• Documentación de dbt Core — docs oficiales de dbt Core 1.x: modelos, contexto Jinja y CLI.
• Materializaciones — cómo se comportan view, table, incremental y ephemeral, y materializaciones custom.
• Sources y ref — declarar sources, freshness checks y cómo ref() construye el DAG.
• dbt_utils y dbt_expectations — paquetes de macros para tests y transformaciones comunes.
• dbt build — corre modelos, tests, snapshots y seeds en orden de DAG.