dbt in Produzione: Data Governance, Semantic Layer e Orchestrazione con Airflow
Architettura dbt enterprise: accesso granulare a modelli con policy di governance, dbt Semantic Layer (MetricFlow) per metriche consistenti che qualunque tool BI può interrogare, integrazione con Apache Airflow per orchestrazione dipendenze cross-pipeline.
dbt in Produzione: Cosa Cambia Rispetto allo Sviluppo
Passare dbt da un progetto di sviluppo a un sistema in produzione usato da decine di analisti e consumer BI richiede tre componenti aggiuntivi che in sviluppo non sono necessari: una strategia di governance per controllare chi vede e modifica cosa, un Semantic Layer per garantire che tutti usino le stesse definizioni di metriche, e un orchestratore robusto per gestire le dipendenze tra dbt e il resto della data platform.
Questo modulo finale della serie affronta ciascuno dei tre temi con approccio pratico: configurazioni YAML reali, codice Airflow funzionante, e pattern MetricFlow pronti all'uso.
Data Governance: Controllo Accessi e Policy
Gruppi e Accesso ai Modelli
dbt 1.5 ha introdotto i gruppi e il sistema di access control
sui modelli: ogni modello può avere un livello di accesso che determina chi può
referenziarlo con 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 stabileVersioning dei Modelli: Contratti Stabili con Consumer
# 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: Garantire lo Schema dei Modelli Pubblici
# 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_nulldbt Semantic Layer: Metriche Consistenti per Tutti i Tool BI
Il Semantic Layer di dbt (basato su MetricFlow) risolve uno dei problemi più frustranti nei data team: metriche calcolate in modo diverso da analisti diversi. “Revenue” in Tableau calcola il lordo, in PowerBI calcola il netto, nel notebook Python include i rimborsi. Chi ha ragione?
Con il Semantic Layer, definisci le metriche una volta in YAML. Ogni tool BI che si connette al Semantic Layer riceve la stessa definizione, la stessa logica, gli stessi numeri.
Semantic Models: Definire Entità e Misure
# 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_amountMetriche: KPI da 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: yearInterrogare il 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()Orchestrazione con Apache Airflow
Airflow è l'orchestratore più usato in ambienti enterprise. L'integrazione
con dbt usa il provider apache-airflow-providers-dbt-cloud (per dbt Cloud)
oppure il Cosmos package (per dbt Core, che traduce il DAG dbt in task Airflow).
Pattern 1: dbt con BashOperator (Semplice)
# 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 è la libreria open-source di Astronomer che traduce ogni modello dbt in un task Airflow separato, rendendo il DAG di Airflow uno specchio perfetto del DAG dbt. Ogni modello appare come task, con dipendenze visualizzate correttamente nell'UI di Airflow.
# 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: Pipeline Cross-Sistema con 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 e Monitoraggio in Produzione
# 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
# }Schema di Governance Enterprise: Tutto Insieme
La visione completa dell'architettura dbt enterprise combina tutti gli elementi visti in questa serie:
# 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.pyChecklist di Maturità per dbt in Produzione
- Governance: gruppi definiti per team, access control su modelli pubblici/protetti/privati, contratti enforced sui modelli critici
- Testing: test generici su ogni primary key, test singolari per regole di business critiche, dbt-expectations per distribuzione e freshness
- CI/CD: Slim CI su ogni PR con GitHub Actions, branch protection, cleanup schema CI automatico
- Documentazione: descrizioni YAML su ogni modello e colonna, doc block per contenuti riutilizzabili, exposures per i consumer
- Monitoring: alerting Slack su failure, source freshness schedulato, health check sui modelli incrementali
- Semantic Layer: metriche core definite una volta, tool BI connessi tramite API MetricFlow
- Orchestrazione: Airflow o Cosmos per le dipendenze cross-pipeline, sensors per le sorgenti upstream
Conclusioni: Fine della Serie dbt
Questa serie ha percorso il ciclo completo di dbt: dall'installazione e setup del primo progetto (modulo 01), alla struttura dei modelli e delle sorgenti (modulo 02), Jinja e macro (modulo 03), materializzazioni (modulo 04), testing (modulo 05), documentazione (modulo 06), confronto Core vs Cloud (modulo 07), CI/CD con Slim CI (modulo 08), incrementali avanzati con microbatch e CDC (modulo 09), fino all'architettura enterprise completa con governance, Semantic Layer e Airflow.
dbt non è solo uno strumento di trasformazione SQL: è un framework che porta le best practice del software engineering - testing, versionamento, documentazione, CI/CD - nel mondo dei dati. La sua forza è rendere queste pratiche accessibili a chi conosce SQL, senza richiedere competenze di sviluppo software profonde.
Il passo successivo naturale dopo questa serie è approfondire l'integrazione di dbt con il resto della modern data stack: Apache Kafka per l'ingestion in tempo reale, Snowflake/BigQuery per ottimizzazioni warehouse-specifiche, e tool BI come Metabase, Superset o Tableau per la visualizzazione dei dati trasformati.
Commenti
Caricamento commenti...
Accedi per commentare