dbt in Production: What Changes Compared to Development

Moving dbt from a development project to a production system used by dozens of analysts and consumer BI requires three additional components that are not necessary in development: one strategy of governance to control who sees and modifies what, a Semantic Layer to ensure everyone uses the same definitions of metrics, and a robust orchestrator to manage dependencies between dbt and the rest of the data platform.

This final module in the series addresses each of the three themes with a practical approach: Real YAML configurations, working Airflow code, and ready-to-use MetricFlow patterns.

Data Governance: Access Control and Policy

Groups and Access to Templates

dbt 1.5 introduced i groups and the system of access control about models: Each model can have an access level that determines who can reference it with ref().

# models/groups.yml
# Definisci i gruppi del progetto

version: 2

groups:
  - name: finance
    owner:
      name: Finance Analytics Team
      email: finance-analytics@company.com

  - name: marketing
    owner:
      name: Marketing Analytics Team
      email: marketing@company.com

  - name: platform
    owner:
      name: Data Platform Team
      email: data-platform@company.com

# models/marts/finance/schema.yml
# Assegna i modelli finance al gruppo finance con accesso protetto

version: 2

models:
  - name: fct_orders
    group: finance
    access: public       # qualsiasi modello del progetto può referenziarlo

  - name: fct_revenue_detailed
    group: finance
    access: protected    # solo modelli dello stesso gruppo possono referenziarlo

  - name: int_order_calculations
    group: finance
    access: private      # solo modelli nella stessa directory possono referenziarlo

  - name: finance_internal_adjustments
    group: finance
    access: private      # dati sensibili: solo il gruppo finance

-- Se il team marketing prova a referenziare un modello privato del team finance:
-- models/marts/marketing/fct_campaigns.sql

SELECT *
FROM {{ ref('finance_internal_adjustments') }}  -- ERRORE in dbt compile!

-- dbt error: Node 'model.my_project.fct_campaigns' attempted to reference
-- 'model.my_project.finance_internal_adjustments', however 'finance_internal_adjustments'
-- is private to the 'finance' group.

-- La soluzione: il team platform espone solo i modelli public come contratto stabile

Model Versioning: Stable Contracts with Consumers

# dbt 1.5+ supporta il versioning dei modelli
# Utile quando modelli pubblici vengono usati da sistemi esterni (API, altri progetti dbt)

# models/marts/v1/schema.yml
version: 2

models:
  - name: fct_orders
    latest_version: 2    # indica quale è la versione corrente

    versions:
      - v: 1
        description: "Versione originale, deprecata. Usa v2."
        config:
          alias: fct_orders_v1
        deprecation_date: 2025-06-01

      - v: 2
        description: "Versione corrente con campo net_revenue e segmentazione cliente."
        columns:
          - include: all
          - name: net_revenue   # colonna aggiunta in v2
          - name: customer_tier # colonna aggiunta in v2

-- Consumer che puntano esplicitamente a una versione:
SELECT * FROM {{ ref('fct_orders', v=1) }}  -- usa la v1 (deprecated)
SELECT * FROM {{ ref('fct_orders', v=2) }}  -- usa la v2 (current)
SELECT * FROM {{ ref('fct_orders') }}        -- usa latest_version (v2)

Contract: Guarantee the Public Model Scheme

# Per modelli pubblici critici, abilita il contract enforcement:
# dbt non può fare dbt run se lo schema non corrisponde esattamente

models:
  - name: fct_orders
    access: public
    group: finance
    config:
      contract:
        enforced: true   # dbt fallisce se le colonne non corrispondono al schema YAML

    columns:
      - name: order_id
        data_type: varchar
        constraints:
          - type: not_null
          - type: unique

      - name: customer_id
        data_type: varchar
        constraints:
          - type: not_null

      - name: total_amount
        data_type: numeric
        constraints:
          - type: not_null

      - name: order_date
        data_type: date
        constraints:
          - type: not_null

dbt Semantic Layer: Consistent Metrics for All BI Tools

Il Semantic Layer of dbt (based on MetricFlow) solves one of the problems most frustrating in data teams: metrics calculated differently by different analysts. “Revenue” in Tableau calculates the gross, in PowerBI it calculates the net, in the notebook Python includes refunds. Who is right?

With the Semantic Layer, you define metrics once in YAML. Every BI tool that you connects to the Semantic Layer receives the same definition, the same logic, the same numbers.

Semantic Models: Defining Entities and Measurements

# models/semantic_models/sem_orders.yml
# Un semantic model è come un "cubo OLAP" dichiarativo

semantic_models:
  - name: orders
    description: "Modello semantico degli ordini completati."
    model: ref('fct_orders')

    # Entità: chiavi per i join semantici
    entities:
      - name: order
        type: primary
        expr: order_id

      - name: customer
        type: foreign
        expr: customer_id

    # Dimensioni: attributi per il raggruppamento
    dimensions:
      - name: order_date
        type: time
        type_params:
          time_granularity: day   # granularità minima: giorno

      - name: status
        type: categorical

      - name: acquisition_channel
        type: categorical

      - name: customer_country
        type: categorical

    # Misure: aggregazioni pre-definite
    measures:
      - name: order_count
        description: "Numero totale di ordini."
        agg: count
        expr: order_id

      - name: gross_revenue
        description: "Ricavo lordo in EUR (prima di sconti)."
        agg: sum
        expr: gross_revenue

      - name: net_revenue
        description: "Ricavo netto in EUR (dopo sconti, prima di rimborsi)."
        agg: sum
        expr: net_revenue

      - name: unique_customers
        description: "Clienti unici con almeno un ordine."
        agg: count_distinct
        expr: customer_id

      - name: avg_order_value
        description: "Valore medio degli ordini."
        agg: average
        expr: total_amount

Metrics: KPIs from Semantic Models

# models/metrics/metrics.yml
# Le metriche referenziano le misure dei semantic models

metrics:
  # Metrica semplice: mapping diretto su una misura
  - name: revenue
    label: "Ricavo Netto"
    description: "Ricavo netto totale in EUR."
    type: simple
    type_params:
      measure: net_revenue

  - name: orders
    label: "Numero Ordini"
    type: simple
    type_params:
      measure: order_count

  # Metrica derivata: operazione su altre metriche
  - name: average_order_value
    label: "Valore Medio Ordine"
    description: "Ricavo netto medio per ordine."
    type: derived
    type_params:
      expr: revenue / orders
      metrics:
        - name: revenue
        - name: orders

  # Metrica ratio: calcolo di proporzione
  - name: organic_revenue_share
    label: "Quota Ricavo Organico"
    description: "Percentuale di ricavo proveniente da canale organico."
    type: ratio
    type_params:
      numerator:
        name: net_revenue
        filter: "{{ Dimension('orders__acquisition_channel') }} = 'organic'"
      denominator:
        name: net_revenue

  # Metrica cumulativa: running total
  - name: cumulative_revenue_ytd
    label: "Ricavo Cumulativo YTD"
    type: cumulative
    type_params:
      measure: net_revenue
      window: unbounded           # dall'inizio dell'anno (con grain=year)
      grain_to_date: year

Query the Semantic Layer

-- Via CLI dbt (dbt Cloud o dbt Core con server semantics):
dbt sl query \
  --metrics revenue,orders,average_order_value \
  --group-by order_date__month,customer_country \
  --where "order_date__month >= '2024-01-01'" \
  --order revenue

-- Output: tabella con colonne:
-- order_date__month | customer_country | revenue | orders | average_order_value

# Via Python con dbt Semantic Layer API (dbt Cloud Enterprise)
from dbt_sl_sdk import SemanticLayerClient

client = SemanticLayerClient(
    environment_id=12345,
    auth_token="my_token"
)

result = client.query(
    metrics=["revenue", "orders", "average_order_value"],
    group_by=["order_date__month", "customer_country"],
    where=[{"filter": "order_date__month >= '2024-01-01'"}],
    order_by=[{"metric": "revenue", "descending": True}]
)

df = result.as_dataframe()

Orchestration with Apache Airflow

Airflow is the most used orchestrator in enterprise environments. Integration with dbt use the provider apache-airflow-providers-dbt-cloud (for dbt Cloud) or the Cosmos package (for dbt Core, which translates the dbt DAG into Airflow tasks).

Pattern 1: dbt with BashOperator (Simple)

# dags/dbt_daily_pipeline.py
# Il pattern più semplice: BashOperator che chiama la CLI dbt

from airflow.decorators import dag, task_group
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from datetime import timedelta

DBT_PROJECT_DIR = '/opt/airflow/dbt'
DBT_PROFILES_DIR = '/opt/airflow/dbt'

default_args = {
    'owner': 'data-platform',
    'depends_on_past': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=10),
    'on_failure_callback': lambda ctx: alert_slack(ctx)
}

@dag(
    dag_id='dbt_daily_production',
    schedule_interval='0 6 * * *',   # 06:00 UTC ogni giorno
    start_date=days_ago(1),
    default_args=default_args,
    catchup=False,
    tags=['dbt', 'production']
)
def dbt_daily():

    source_freshness = BashOperator(
        task_id='dbt_source_freshness',
        bash_command=f'cd {DBT_PROJECT_DIR} && dbt source freshness --target prod',
        env={'DBT_PROFILES_DIR': DBT_PROFILES_DIR}
    )

    run_staging = BashOperator(
        task_id='dbt_run_staging',
        bash_command=f'cd {DBT_PROJECT_DIR} && dbt run --select tag:staging --target prod',
        env={'DBT_PROFILES_DIR': DBT_PROFILES_DIR}
    )

    test_staging = BashOperator(
        task_id='dbt_test_staging',
        bash_command=f'cd {DBT_PROJECT_DIR} && dbt test --select tag:staging --target prod',
        env={'DBT_PROFILES_DIR': DBT_PROFILES_DIR}
    )

    run_marts = BashOperator(
        task_id='dbt_run_marts',
        bash_command=f'cd {DBT_PROJECT_DIR} && dbt run --select tag:daily --exclude tag:staging --target prod',
        env={'DBT_PROFILES_DIR': DBT_PROFILES_DIR}
    )

    test_marts = BashOperator(
        task_id='dbt_test_marts',
        bash_command=f'cd {DBT_PROJECT_DIR} && dbt test --select tag:daily --exclude tag:staging --target prod',
        env={'DBT_PROFILES_DIR': DBT_PROFILES_DIR}
    )

    run_snapshots = BashOperator(
        task_id='dbt_snapshots',
        bash_command=f'cd {DBT_PROJECT_DIR} && dbt snapshot --target prod',
        env={'DBT_PROFILES_DIR': DBT_PROFILES_DIR}
    )

    # Definizione delle dipendenze
    source_freshness >> run_staging >> test_staging >> run_marts >> test_marts
    run_staging >> run_snapshots  # snapshots in parallelo ai marts

dag = dbt_daily()

Pattern 2: Astronomer Cosmos (DAG dbt → Task Airflow)

Cosmos is Astronomer's open-source library that translates every dbt model in a separate Airflow task, making the Airflow DAG a perfect mirror of the DAG dbt. Each model appears as a task, with dependencies displayed correctly in the Airflow UI.

# dags/dbt_cosmos_dag.py
# Cosmos: ogni modello dbt diventa un task Airflow separato

from airflow.decorators import dag
from airflow.utils.dates import days_ago
from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping

profile_config = ProfileConfig(
    profile_name="my_project",
    target_name="prod",
    profile_mapping=SnowflakeUserPasswordProfileMapping(
        conn_id="snowflake_prod",
        profile_args={
            "database": "analytics",
            "schema": "marts",
            "warehouse": "compute_wh_m"
        }
    )
)

@dag(
    dag_id='dbt_cosmos_pipeline',
    schedule_interval='0 6 * * *',
    start_date=days_ago(1),
    catchup=False
)
def cosmos_dbt_dag():
    dbt_project = DbtDag(
        project_config=ProjectConfig(
            dbt_project_path='/opt/airflow/dbt',
        ),
        profile_config=profile_config,
        execution_config=ExecutionConfig(
            dbt_executable_path='/usr/local/bin/dbt',
        ),
        # Seleziona solo i modelli taggati 'daily'
        select=['tag:daily'],
        schedule_interval=None,   # il DAG padre gestisce lo schedule
        dag_id='dbt_models_daily'
    )
    return dbt_project

dag = cosmos_dbt_dag()

Pattern 3: Cross-System Pipeline with Sensors

# dags/full_data_platform_pipeline.py
# Pattern reale: dbt aspetta il completamento dell'ingestion (Fivetran/Airbyte)

from airflow.decorators import dag, task
from airflow.operators.bash import BashOperator
from airflow.providers.http.sensors.http import HttpSensor
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.dates import days_ago
from datetime import timedelta

@dag(
    dag_id='full_data_platform',
    schedule_interval='0 5 * * *',
    start_date=days_ago(1),
    catchup=False
)
def full_pipeline():

    # 1. Aspetta il completamento della sync Fivetran (webhook o polling)
    wait_for_fivetran = HttpSensor(
        task_id='wait_fivetran_sync',
        http_conn_id='fivetran_api',
        endpoint='/v1/connectors/abc123/last-sync',
        response_check=lambda response: response.json()['data']['status']['sync_state'] == 'ready',
        poke_interval=300,    # controlla ogni 5 minuti
        timeout=7200,         # timeout dopo 2 ore
        mode='reschedule'     # libera il worker tra i check
    )

    # 2. Aspetta il DAG di ingestion Kafka se esiste
    wait_for_kafka_ingestion = ExternalTaskSensor(
        task_id='wait_kafka_ingestion',
        external_dag_id='kafka_events_ingestion',
        external_task_id='load_to_warehouse',
        timeout=3600,
        mode='reschedule'
    )

    # 3. dbt source freshness (verifica che i dati siano davvero arrivati)
    source_freshness = BashOperator(
        task_id='dbt_source_freshness',
        bash_command='cd /opt/airflow/dbt && dbt source freshness --target prod'
    )

    # 4. dbt build completo
    dbt_build = BashOperator(
        task_id='dbt_build',
        bash_command='cd /opt/airflow/dbt && dbt build --target prod --select tag:daily'
    )

    # 5. Notifica downstream: BI refresh, ML retraining, ecc.
    @task
    def notify_downstream():
        import requests
        # Trigger refresh Tableau
        requests.post('https://tableau.company.com/api/refresh',
                     headers={'X-Tableau-Auth': '{{ var.value.tableau_token }}'})
        # Notifica il team su Slack
        requests.post('{{ var.value.slack_webhook }}',
                     json={'text': 'Pipeline giornaliera completata - dati aggiornati'})

    [wait_for_fivetran, wait_for_kafka_ingestion] >> source_freshness >> dbt_build >> notify_downstream()

dag = full_pipeline()

Alerting and Monitoring in Production

# utils/dbt_alerts.py
# Funzione di alerting riutilizzabile per i DAG Airflow

import requests
from airflow.models import Variable

def alert_slack_on_failure(context):
    """
    Invia notifica Slack quando un task dbt fallisce.
    Da usare come on_failure_callback nei DAG Airflow.
    """
    dag_id = context['dag'].dag_id
    task_id = context['task_instance'].task_id
    run_id = context['run_id']
    log_url = context['task_instance'].log_url

    message = {
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f":red_circle: *dbt Task Fallito*\n"
                            f"*DAG:* `{dag_id}`\n"
                            f"*Task:* `{task_id}`\n"
                            f"*Run ID:* `{run_id}`\n"
                            f"*Log:* <{log_url}|Visualizza log>"
                }
            }
        ]
    }

    slack_webhook = Variable.get('SLACK_WEBHOOK_DATA_ALERTS')
    requests.post(slack_webhook, json=message)


# Utilizzo nel DAG:
# default_args = {
#     'on_failure_callback': alert_slack_on_failure
# }

Enterprise Governance Framework: All Together

The complete vision of the dbt enterprise architecture combines all the elements seen in this series:

# Struttura del progetto dbt enterprise completo
#
# my_dbt_project/
# ├── models/
# │   ├── staging/               # view, test generici, freshness check
# │   ├── intermediate/          # ephemeral, logica di trasformazione
# │   ├── marts/
# │   │   ├── finance/           # gruppo: finance, access: public/protected
# │   │   ├── marketing/         # gruppo: marketing, access: public
# │   │   └── platform/          # gruppo: platform (modelli shared)
# │   └── semantic_models/       # MetricFlow semantic models
# ├── metrics/                   # definizioni metriche
# ├── snapshots/                 # SCD-2 snapshots
# ├── macros/                    # macro riutilizzabili
# ├── tests/                     # test singolari e generici custom
# ├── analyses/                  # query ad-hoc versionata
# ├── seeds/                     # dati statici di riferimento
# ├── packages.yml               # dbt-utils, dbt-expectations
# ├── dbt_project.yml            # configurazione progetto + materializzazioni
# ├── profiles.yml               # connessioni warehouse (escluso da git con .gitignore per prod)
# ├── requirements.txt           # versioni pinned
# └── .github/workflows/
#     ├── dbt-ci.yml             # Slim CI su PR
#     └── dbt-prod.yml           # Deploy + artifact upload

# Airflow DAGs (repo separato o sottocartella):
# ├── dags/
# │   ├── dbt_daily_pipeline.py
# │   └── full_data_platform.py
# └── utils/
#     └── dbt_alerts.py

Conclusions: End of the dbt Series

This series has covered the complete cycle of dbt: from the installation and setup of the first one project (module 01), the structure of models and sources (module 02), Jinja and macros (module 03), materializations (module 04), testing (module 05), documentation (module 06), Core vs Cloud comparison (module 07), CI/CD with Slim CI (module 08), advanced incrementals with microbatch and CDC (module 09), up to complete enterprise architecture with governance, Semantic Layer and Airflow.

dbt isn't just a SQL transformation tool - it's a framework that brings software engineering best practices - testing, versioning, documentation, CI/CD - in the world of data. Its strength is making these practices accessible to those who know SQL, without requiring deep software development skills.

The natural next step after this series is to delve deeper into dbt integration with the rest of the modern data stack: Apache Kafka for real-time ingestion, Snowflake/BigQuery for warehouse-specific optimizations, and BI tools like Metabase, Superset or Tableau for visualizing transformed data.