Automatiza ELT con dbt: Guía de orquestación para CTOs

En la empresa moderna, el almacén de datos (DW) se ha convertido en el sistema nervioso central, el cerebro estratégico que impulsa el análisis, la inteligencia artificial y la toma de decisiones. Sin embargo, para muchas organizaciones, este activo crítico está limitado por procesos de transformación frágiles, opacos y manuales. Los procesos ETL tradicionales son lentos, difíciles de mantener y, fundamentalmente, carecen de las mejores prácticas de ingeniería de software que exigimos de nuestro código de aplicación.

Esto es donde dbt (herramienta de construcción de datos) cambia fundamentalmente el paradigma. Al pasar a un modelo ELT (Extracción, Carga, Transformación), dbt introduce los principios de la ingeniería de análisis (modularidad, control de versiones, pruebas y CI/CD) en la capa de transformación (T) que reside dentro de tu almacén de datos.

Este artículo no es una descripción general. Es un plan técnico y práctico para directores de tecnología (CTOs) e ingenieros senior, que les muestra cómo implementar un flujo de trabajo dbt completamente automatizado y de producción. Nos centraremos en construir un robusto sistema de CI/CD y una estrategia de orquestación consciente del estado para llevar su práctica de datos de scripts improvisados a una fábrica de datos de alta velocidad y confiable.

Servicios de Ingeniería de Productos

Trabaje con nuestros gestores de proyectos, ingenieros de software y testers de control de calidad para desarrollar su nuevo producto de software personalizado o para apoyar su flujo de trabajo actual, siguiendo metodologías Agile, DevOps y Lean.

Build with 4Geeks

El argumento de 'Por qué dbt?', para el liderazgo técnico

Antes de profundizar en la automatización, debemos tener claro qué estamos automatizando. dbt es un compilador y un marco de ejecución para la lógica de sus datos. No extrae ni carga datos; gestiona magistralmente la transformación de los datos brutos que ya están almacenados en su almacén (p. ej., Snowflake, BigQuery, Redshift, Databricks).

Para un líder técnico, el valor de dbt es evidente:

• Es simplemente SQL (y Jinja): Aprovecha las habilidades existentes de SQL de tu equipo de análisis, envolviéndolas en un potente motor de plantillas (Jinja) para la modularidad y los principios DRY (Don't Repeat Yourself).
• Flujo de trabajo nativo de Git: Cada cambio en un modelo de datos es un commit. Cada nueva función es una rama. Cada lanzamiento a producción es una fusión a main. Esto proporciona auditabilidad, colaboración y capacidades de reversión.
• Pruebas integradas: dbt proporciona un marco de primer nivel para las afirmaciones de calidad de datos. Puedes definir pruebas (por ejemplo, not_null, unique, accepted_values, basadas en SQL personalizadas) en archivos YAML simples, que dbt ejecuta contra tu almacén.
• Gestión de dependencias y trazabilidad: La función ref() de dbt es su característica más crítica. En lugar de codificar nombres de tablas, referencias a otros modelos. dbt utiliza esto para inferir automáticamente un Gráfico Acíclico Dirigido (DAG) de dependencias. Sabe exactamente qué orden seguir para construir los modelos y puede generar un gráfico completo de trazabilidad de datos para todo tu almacén (documentación de dbt).

En resumen, dbt trata la lógica de transformación de tus datos como un producto de software, lo cual es un requisito previo para la automatización significativa.

Implementar un pipeline de CI/CD para dbt

La regla fundamental de una configuración de producción de dbt es:Ningún desarrollador, analista o ingeniero debería ejecutar dbt run directamente contra el entorno de producción desde su portátil. Todas las modificaciones deben ser revisadas por pares y validadas a través de una canalización de CI (Integración Continua) automatizada.

El objetivo de CI es responder a una pregunta para cada Solicitud de Incorporación (PR): "¿Esta modificación propuesta causa algún problema?"

La estrategia más efectiva para esto es"Slim CI", que aprovecha las capacidades de estado de dbt.únicamente los modelos que han sido modificados y sus dependencias asociadas.

Ejemplo de trabajo de CI paso a paso (GitHub Actions)

Esta acción se activa en cada solicitud de extracción contra la rama principal . Crea los modelos modificados en un esquema temporal y aislado dentro de tu almacén de datos.

Requisito: Esta estrategia requiere que almacene su archivo "manifest.json" en una ubicación accesible (como S3, GCS o Azure Blob Storage) después de cada ejecución de producción exitosa. Este archivo "manifest.json" es un mapa de su entorno de producción, que dbt utiliza para calcular la diferencia de "estado".manifest.jsonarchivo en una ubicación accesible (como S3, GCS o Azure Blob Storage) después de cada ejecución exitosa. Este manifiesto es un mapa de su entorno de producción, que dbt utiliza para calcular la diferencia de "estado".

Aquí tienes un archivo .github/workflows/dbt-ci.yml detallado:

name: dbt CI (Slim)

on:
  pull_request:
    branches:
      - main

env:
  DBT_PROFILES_DIR: . # Tell dbt where to find profiles.yml
  # Store all warehouse credentials as GitHub Secrets
  DBT_USER: ${{ secrets.DBT_USER_CI }}
  DBT_PASSWORD: ${{ secrets.DBT_PASSWORD_CI }}
  DBT_ACCOUNT: ${{ secrets.DBT_ACCOUNT }}
  DBT_ROLE: ${{ secrets.DBT_ROLE_CI }}
  DBT_WAREHOUSE: ${{ secrets.DBT_WAREHOUSE_CI }}
  DBT_DATABASE: ${{ secrets.DBT_DATABASE_PROD }} # Use prod database

jobs:
  run_dbt_slim_ci:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Download production manifest
        # This step downloads the manifest from your artifact storage.
        # This example uses a GCS bucket.
        uses: 'google-github-actions/auth@v1'
        with:
          credentials_json: '${{ secrets.GCS_SA_KEY }}'
        env:
          GCP_BUCKET: "your-dbt-artifacts-bucket"
          
      - name: 'Download Manifest'
        uses: 'google-github-actions/storage-transfer@v1'
        with:
          source: 'gs://${{ env.GCP_BUCKET }}/prod/manifest.json'
          destination: './prod-manifest'
        # Continue if no manifest exists (e.g., first run)
        continue-on-error: true

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'

      - name: Install dbt and adapter
        run: |
          pip install dbt-snowflake==1.8.0 # Use your specific adapter and version
          dbt --version

      - name: Configure dbt profile (for CI)
        # Dynamically create profiles.yml from secrets
        # This builds into a unique, temporary schema for this PR
        run: |
          echo "default:" > profiles.yml
          echo "  target: ci" >> profiles.yml
          echo "  outputs:" >> profiles.yml
          echo "    ci:" >> profiles.yml
          echo "      type: snowflake"
          echo "      account: $DBT_ACCOUNT"
          echo "      user: $DBT_USER"
          echo "      password: $DBT_PASSWORD"
          echo "      role: $DBT_ROLE"
          echo "      warehouse: $DBT_WAREHOUSE"
          echo "      database: $DBT_DATABASE"
          # CRITICAL: Isolate the build
          echo "      schema: dbt_ci_pr_${{ github.event.pull_request.number }}"
          echo "      threads: 4"
          echo "      client_session_keep_alive: False"

      - name: Install dbt dependencies
        run: dbt deps

      - name: Run and test only modified models (Slim CI)
        # 'state:modified+' selector finds all modified models AND
        # all downstream models that depend on them.
        # --state flag points to the downloaded production manifest.
        run: |
          dbt build --select state:modified+ --state ./prod-manifest
        # If no manifest was found, fall back to a full build (slower)
        continue-on-error: false 
        # TODO: Add a fallback step here if the first run fails
        # e.g., if: failure()
        #       run: dbt build --select state:modified+
        
      - name: Clean up temporary schema
        # This runs whether the build succeeds or fails
        if: always()
        run: |
          dbt run-operation drop_pr_schema --args '{schema_name: "dbt_ci_pr_${{ github.event.pull_request.number }}"}'

Al fusionar a main

Cuando una solicitud de extracción (PR) se fusiona, la rama principal representa tu artefacto de despliegue. Una tarea en la rama principal no debería desplegarse en producción. En cambio, debería:

1. Realiza una última construcción de dbt contra un entorno de Preparación dedicado (una copia sin copia del entorno de producción es ideal).
2. Si tiene éxito, genera y guarda el archivo manifest.json. Este archivo ahora representa el estado de la versión principal y se utilizará tanto por tu orquestador de producción como por futuras ejecuciones de CI.

Servicios de Ingeniería de Productos

Trabaje con nuestros gestores de proyectos, ingenieros de software y probadores de calidad internos para desarrollar su nuevo producto de software personalizado o para apoyar su flujo de trabajo actual, siguiendo las metodologías Agile, DevOps y Lean.

Build with 4Geeks

Orquestación de la producción (La parte 'CD')

CI construye y prueba el código. Un Orchestrador construye y prueba los datos según un horario. Tu herramienta de orquestación (p. ej., Airflow, Dagster, Prefect, dbt Cloud) es responsable de ejecutar tu proyecto dbt en producción.

Principios clave:

1. Separación de Responsabilidades: El DAG del orquestador debe ser simple. Obtiene la última rama principal, obtiene el último archivo manifest.json, y ejecuta comandos de dbt.
2. Ejecuciones Conociendo el Estado: Las ejecuciones de producción deben también utilizar el manifest.json.dbt run --state ./ --manifest ./manifest.json, lo que ahorra un tiempo de cálculo significativo.
3. Verificaciones de Calidad de Datos: La tubería debe estar controlada por verificaciones de calidad de datos.

Ejemplo: Orquestación de la producción con Airflow

Este ejemplo asume que estás utilizando Airflow con el BashOperator o DockerOperator para ejecutar comandos dbt.

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

# Define default arguments
default_args = {
    'owner': 'data_engineering',
    'depends_on_past': False,
    'start_date': days_ago(1),
    'retries': 1,
}

with DAG(
    dag_id="dbt_production_pipeline",
    default_args=default_args,
    schedule_interval="0 6 * * *", # 6 AM daily
    catchup=False,
    tags=["dbt", "production"],
) as dag:

    # 1. Pull the latest dbt project code
    git_pull = BashOperator(
        task_id="git_pull",
        bash_command="cd /path/to/dbt-project && git pull origin main",
    )

    # 2. Download the latest production manifest from storage
    #    (This step is analogous to the CI step)
    download_manifest = BashOperator(
        task_id="download_manifest",
        bash_command="gcloud storage cp gs://your-dbt-artifacts-bucket/prod/manifest.json /path/to/dbt-project/prod-artifacts/",
    )

    # 3. Check that raw source data has arrived
    check_source_freshness = BashOperator(
        task_id="check_source_freshness",
        bash_command="cd /path/to/dbt-project && dbt source freshness",
        # If this fails, the whole pipeline fails and alerts
    )

    # 4. Run dbt models (state-aware)
    run_dbt_models = BashOperator(
        task_id="run_dbt_models",
        bash_command=(
            "cd /path/to/dbt-project && "
            "dbt run --state ./prod-artifacts --manifest ./prod-artifacts/manifest.json"
        ),
    )

    # 5. Test dbt models
    test_dbt_models = BashOperator(
        task_id="test_dbt_models",
        bash_command="cd /path/to/dbt-project && dbt test",
    )

    # 6. Generate docs and the NEW manifest
    generate_artifacts = BashOperator(
        task_id="generate_artifacts",
        bash_command="cd /path/to/dbt-project && dbt docs generate",
    )

    # 7. Upload the new manifest for the next CI/prod run
    upload_new_manifest = BashOperator(
        task_id="upload_new_manifest",
        bash_command=(
            "cd /path/to/dbt-project && "
            "gcloud storage cp ./target/manifest.json gs://your-dbt-artifacts-bucket/prod/manifest.json"
        ),
    )

    # Define the DAG's execution order
    (
        git_pull 
        >> download_manifest 
        >> check_source_freshness 
        >> run_dbt_models 
        >> test_dbt_models 
        >> generate_artifacts 
        >> upload_new_manifest
    )

Nota: Para una configuración verdaderamente robusta, utilizarías el KubernetesPodOperator o el DockerOperator para ejecutar cada paso en un contenedor aislado que contenga el código y las credenciales de tu proyecto dbt.

Estrategia avanzada: Implementaciones sin interrupciones

Para datos de misión crítica, no puede permitirse que un entorno dbt falle a mitad de proceso, dejando sus tablas de producción en un estado inoperable. La solución es una estrategia de implementación azul/verde.

1. Su orquestador (por ejemplo, Airflow) no escribe en la PROD base de datos.
2. En cambio, utiliza dbt clone (o la copia ZERO-COPY CLONE de Snowflake) para crear una copia perfecta, solo con metadatos: PROD_CLONE.
3. Toda la pipeline de dbt y las pruebas de dbt se ejecutan contra esta PROD_CLONE base de datos.
4. Si todos los pasos tienen éxito, el "despliegue" es un solo comando atómico de SWAP que intercambia PROD y PROD_CLONE.
5. Sus herramientas de BI y los usuarios, que apuntaban a PROD, ahora apuntan sin problemas a los nuevos datos, que han sido completamente probados. La antigua PROD base de datos (ahora llamada PROD_CLONE) puede ser eliminada.

Este enfoque garantiza que tus consumidores de datos nunca vean una tabla incompleta y que todas las comprobaciones de calidad de los datos hayan superado, antes de que los datos se sirvan.

Servicios de Ingeniería de Productos

Trabaje con nuestros gestores de proyectos, ingenieros de software y probadores de calidad, para desarrollar su nuevo producto de software personalizado o para apoyar su flujo de trabajo actual, siguiendo metodologías Agile, DevOps y Lean.

Build with 4Geeks

Finalmente

Automatizar tu almacén de datos con dbt no se trata solo de programar un comando de ejecución de dbt, sino de adoptar una metodología completa de ingeniería de análisis que considera las transformaciones de tus datos como un servicio de software de producción.

Al implementar unacadena de integración continuadelgada, le da a su equipo la confianza para desarrollar funciones rápidamente, sabiendo que cada cambio se valida automáticamente contra un entorno similar al de producción.

Al crear unaplataforma de orquestación que tiene en cuenta el estado, se crea una fábrica de datos resistente, eficiente y monitorizable que puede recuperarse de los fallos y garantizar la calidad de los datos.

Esta capa de automatización es la infraestructura fundamental que permite aprovechar al máximo el potencial de tu equipo de datos. Permite que se centren en el análisis de alto valor y en la creación de productos de datos que impulsarán el crecimiento de tu empresa, en lugar de centrarse en tareas rutinarias y en la validación manual.

Preguntas frecuentes