DataOps y testing de SQL con dbt y testcontainers

El SQL ha pasado de ser código que se escribe ad hoc en una pestaña de pgAdmin a ser código de producción que merece las mismas prácticas que el resto: control de versiones, testing automatizado, lint, documentación y CI/CD. Esta disciplina se denomina DataOps y combina herramientas modernas con prácticas de software engineering.

El proyecto SQL como código fuente

Un proyecto SQL profesional tiene una estructura semejante a cualquier proyecto de software:

mi-proyecto-data/
├── models/                  # Vistas y transformaciones (dbt)
│   ├── staging/
│   ├── intermediate/
│   └── marts/
├── tests/                   # Tests SQL declarativos
├── macros/                  # Funciones SQL reutilizables (Jinja)
├── snapshots/               # Snapshots para SCD2
├── seeds/                   # CSV con datos de referencia
├── migrations/              # Cambios de esquema versionados (Flyway, Alembic)
├── tests-integracion/       # Tests Python/Java con testcontainers
├── dbt_project.yml
├── profiles.yml
├── .sqlfluff
├── .pre-commit-config.yaml
└── README.md

Esto se versiona en Git. Cada cambio pasa por revisión, CI, despliegue gradual.

dbt: el estándar para transformaciones

dbt (data build tool) es la herramienta más extendida para definir transformaciones SQL como modelos declarativos. Cada modelo es un SELECT que dbt convierte en CREATE TABLE AS o CREATE VIEW:

-- models/marts/pedidos_mensuales.sql
{{ config(materialized='table', schema='analytics') }}

WITH pedidos AS (
    SELECT
        DATE_TRUNC('month', creado_en) AS mes,
        cliente_id,
        SUM(total) AS facturacion,
        COUNT(*) AS num_pedidos
    FROM {{ source('produccion', 'pedidos') }}
    WHERE estado = 'PAGADO'
    GROUP BY 1, 2
)
SELECT * FROM pedidos

La sintaxis {{ source('produccion', 'pedidos') }} referencia una fuente declarada en sources.yml. dbt genera el FROM correcto según el entorno (dev, staging, prod) sin hardcodear esquemas.

# models/sources.yml
sources:
  - name: produccion
    schema: public
    tables:
      - name: pedidos
        columns:
          - name: id
            tests:
              - unique
              - not_null
          - name: cliente_id
            tests:
              - relationships:
                  to: source('produccion', 'clientes')
                  field: id

Los tests declarativos se ejecutan automáticamente:

dbt test
# unique on pedidos.id ........... PASS
# not_null on pedidos.id ......... PASS
# relationships on pedidos.cliente_id ... PASS

dbt convierte el SQL en un grafo dirigido. Conoce las dependencias entre modelos y los ejecuta en el orden correcto: si un modelo depende de otro, se construye después.

Tests adicionales con dbt

Más allá de los tests genéricos, dbt permite tests personalizados:

-- tests/no_pedidos_negativos.sql
SELECT id
FROM {{ ref('pedidos_mensuales') }}
WHERE facturacion < 0

Si la consulta devuelve filas, el test falla. Es la forma más simple de codificar invariantes de negocio.

Tests más sofisticados se definen como macros:

-- macros/test_freshness.sql
{% test freshness_max_hours(model, column_name, max_hours) %}
SELECT MAX({{ column_name }}) AS ultima_fecha
FROM {{ model }}
HAVING NOW() - MAX({{ column_name }}) > INTERVAL '{{ max_hours }} hours'
{% endtest %}

Y se aplican declarativamente:

models:
  - name: eventos_recientes
    columns:
      - name: ocurrido_en
        tests:
          - freshness_max_hours:
              max_hours: 24

sqlfluff: linter de SQL

sqlfluff impone convenciones de estilo: keywords en mayúsculas, indentación consistente, alias coherentes. Configuración mínima en .sqlfluff:

[sqlfluff]
dialect = postgres
templater = jinja
exclude_rules = L034

[sqlfluff:rules]
max_line_length = 120
capitalisation_policy = upper

[sqlfluff:indentation]
indented_joins = false
indented_using_on = false

Ejecutar:

sqlfluff lint models/
sqlfluff fix models/  # arregla automaticamente lo que puede

Integrado como pre-commit hook:

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/sqlfluff/sqlfluff
    rev: 3.0.0
    hooks:
      - id: sqlfluff-lint
      - id: sqlfluff-fix

Un equipo con linter es un equipo que no discute de estilo en code reviews. La revisión se concentra en lógica y casos límite.

Tests de integración con testcontainers

dbt y sqlfluff cubren la calidad de las queries. Las migraciones de esquema y la lógica de aplicación requieren tests con una base de datos real. Testcontainers arranca un PostgreSQL efímero en Docker para cada test:

# tests/test_pedidos_repository.py
import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine

@pytest.fixture(scope="session")
def postgres_engine():
    with PostgresContainer("postgres:16") as postgres:
        engine = create_engine(postgres.get_connection_url())
        run_migrations(engine)
        yield engine

def test_pedido_se_persiste_y_se_recupera(postgres_engine):
    repo = PedidoRepository(postgres_engine)
    pedido = Pedido(cliente_id=1, total=99.99)
    repo.save(pedido)

    recuperado = repo.get_by_id(pedido.id)
    assert recuperado.total == 99.99

El equivalente en Java con JUnit 5:

@Testcontainers
class PedidoRepositoryTest {

    @Container
    PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16");

    @Test
    void pedidoSePersisteYRecupera() {
        DataSource ds = createDataSource(postgres);
        runMigrations(ds);

        PedidoRepository repo = new PedidoRepository(ds);
        Pedido pedido = new Pedido(1L, BigDecimal.valueOf(99.99));
        repo.save(pedido);

        Pedido recuperado = repo.getById(pedido.getId());
        assertEquals(BigDecimal.valueOf(99.99), recuperado.getTotal());
    }
}

Cada test arranca su propio contenedor (o reutiliza uno por sesión). El estado nunca se contamina entre tests porque la base de datos es efímera.

Snapshot tests para detectar regresiones

Cuando una query genera un resultado complejo, los snapshot tests lo capturan en un fichero y comparan en futuros runs:

def test_informe_ventas_diarias(postgres_engine, snapshot):
    cargar_datos_fixture(postgres_engine)

    resultado = ejecutar_query(postgres_engine, "informe_ventas_diarias.sql")

    snapshot.assert_match(resultado.to_csv(), "informe_ventas_diarias.csv")

La primera ejecución crea el snapshot. Las siguientes comparan; si algo cambia, el test falla y muestra el diff. Útil para detectar regresiones por cambios accidentales en una transformación.

Pipeline CI completa

Una pipeline de GitHub Actions o GitLab CI típica:

# .github/workflows/data.yml
name: Data CI
on: [pull_request]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: postgres
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
        ports: ["5432:5432"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
      - run: pip install dbt-postgres sqlfluff pytest testcontainers
      - run: sqlfluff lint models/
      - run: dbt deps && dbt seed && dbt run && dbt test
      - run: pytest tests-integracion/

Cada PR ejecuta:

• Lint con sqlfluff.
• Carga de datos de referencia con dbt seed.
• Construcción de los modelos con dbt run.
• Tests declarativos con dbt test.
• Tests de integración con testcontainers.

Si algún paso falla, el merge se bloquea. La calidad se mantiene de forma automática, sin depender de la disciplina individual.

Documentación generada

dbt genera documentación automáticamente a partir del proyecto, incluyendo el grafo de dependencias entre modelos y los tests de cada columna:

dbt docs generate
dbt docs serve  # arranca un servidor local con la documentacion

La documentación se publica como sitio estático y sirve de catalogo de datos consumible por el equipo de negocio.

Convención: feature branches y trunk-based development

El flujo recomendado:

• Feature branches cortas (1-3 días) para cada cambio.
• Code review obligatoria, al menos un revisor.
• CI verde como prerequisito para merge.
• Trunk (main) siempre desplegable.
• Despliegue automático a staging, manual o automático a producción tras validación.
• Backwards compatibility: cambios de esquema con expand-contract (visto en lección anterior).

flowchart LR
    A[Pull Request] --> B[CI: sqlfluff]
    B --> C[CI: dbt run y test]
    C --> D[CI: testcontainers]
    D --> E{Todo verde?}
    E -->|Si| F[Merge a main]
    E -->|No| G[Bloqueado]
    F --> H[Despliegue staging]
    H --> I[Validacion manual o automatica]
    I --> J[Despliegue produccion]

DataOps no es solo herramientas: es la mentalidad de tratar el SQL como software de primera clase. Equipos que la adoptan reducen incidentes, aceleran cambios y construyen plataformas de datos en las que el negocio confía. La inversión inicial en setup se amortiza en pocos meses.