dbt (data build tool) est l'outil standard de l'analytics engineering. Il permet d'appliquer les principes du génie logiciel (tests, docs, CI/CD, versioning) aux transformations de données dans le warehouse.
Concepts clés :
Models : fichiers SQL SELECT dans /models/ → tables/vues dans le warehouse.
Tests : assertions de qualité sur les données (unique, not_null, custom).
Documentation : générée automatiquement à partir des descriptions et des tests.
# models/staging/sources.ymlversion:2sources:-name:bronzeschema:bronzedatabase:dw_proddescription:"Données brutes importées depuis le lac de données"tables:-name:sensor_readingsdescription:"Relevés bruts des capteurs"columns:-name:machine_iddescription:"Identifiant de la machine"tests:-not_null-name:temperaturedescription:"Température en °C"-name:timestampdescription:"Horodatage de la mesure"loaded_at_field:_ingested_atfreshness:warn_after: {count:30, period:minute}
error_after: {count:60, period:minute}
-name:machinesdescription:"Référentiel des machines"columns:-name:machine_idtests: [unique, not_null]
2.2 Staging Model (Nettoyage)
-- models/staging/stg_sensors.sql
{{ config(
materialized='view',
bind=False,
tags=['staging', 'hourly']
) }}
SELECT
machine_id::VARCHAR(20) AS machine_id,
temperature::DECIMAL(6,2) AS temperature,
pression::DECIMAL(6,2) AS pression,
timestamp::TIMESTAMPAStimestamp,
_ingested_at::TIMESTAMPAS ingested_at,
CASEWHEN temperature >100THEN'CRITIQUE'WHEN temperature >80THEN'WARNING'ELSE'NORMAL'ENDAS alerte
FROM {{ source('bronze', 'sensor_readings') }}
WHEREtimestampISNOT NULLAND temperature ISNOT NULL
2.3 Dimension Model
-- models/marts/core/dim_machine.sql
{{ config(
materialized='table',
unique_key='machine_sk',
tags=['dimension', 'daily']
) }}
WITH machines AS (
SELECT*FROM {{ ref('stg_machines') }}
),
finalAS (
SELECT
{{ dbt_utils.generate_surrogate_key(['machine_id']) }} AS machine_sk,
machine_id,
machine_type,
site,
zone,
status,
install_date,
CURRENT_TIMESTAMPAS dbt_loaded_at
FROM machines
WHERE is_current =TRUE
)
SELECT*FROMfinal
2.4 Fact Model
-- models/marts/core/fact_sensor_readings.sql
{{ config(
materialized='incremental',
unique_key=['machine_sk', 'timestamp'],
incremental_strategy='merge',
partition_by={field: 'timestamp', data_type: 'date', granularity: 'day'},
tags=['fact', 'real_time']
) }}
WITH sensors AS (
SELECT*FROM {{ ref('stg_sensors') }}
{% if is_incremental() %}
WHEREtimestamp> (SELECTMAX(timestamp) FROM {{ this }})
{% endif %}
),
machines AS (
SELECT machine_sk, machine_id
FROM {{ ref('dim_machine') }}
WHERE is_current =TRUE
)
SELECT
{{ dbt_utils.generate_surrogate_key(['m.machine_sk', 's.timestamp']) }} AS reading_sk,
m.machine_sk,
s.temperature,
s.pression,
s.alerte,
s.timestamp,
s.ingested_at
FROM sensors s
LEFTJOIN machines m ON s.machine_id = m.machine_id
2.5 Mart de Reporting
-- models/marts/reporting/kpi_daily.sql
{{ config(
materialized='table',
tags=['reporting', 'daily']
) }}
SELECT
d.date,
m.site,
m.zone,
COUNT(DISTINCT f.machine_sk) AS machines_actives,
AVG(f.temperature) AS temp_moyenne,
MAX(f.temperature) AS temp_max,
STDDEV(f.temperature) AS temp_std,
COUNT(*) AS nb_releves,
SUM(CASEWHEN f.alerte ='CRITIQUE'THEN1ELSE0END) AS nb_alertes_critiques
FROM {{ ref('fact_sensor_readings') }} f
LEFTJOIN {{ ref('dim_machine') }} m ON f.machine_sk = m.machine_sk
LEFTJOIN {{ ref('dim_date') }} d ONDATE(f.timestamp) = d.date
GROUPBY d.date, m.site, m.zone
ORDERBY d.date, m.site
-- tests/assert_positive_temperature.sql-- Vérifie qu'il n'y a pas de température négativeSELECT machine_id, temperature, timestampFROM {{ ref('stg_sensors') }}
WHERE temperature <-50
# models/marts/core/schema.yml (suite)models:-name:fact_sensor_readingsdescription:>
Table de faits des relevés de capteurs. Chaque ligne représente
une mesure unique d'une machine à un instant donné.
docs:show:truenode_color:'#2c7fb8'columns:-name:reading_skdescription:"Clé de substitution unique (surrogate key)"-name:temperaturedescription:"Température mesurée en degrés Celsius"tests:-not_null
# .github/workflows/dbt_ci.ymlname:dbtCIon: [push]
jobs:dbt-test:runs-on:ubuntu-lateststeps:-uses:actions/checkout@v4-name:Installdbtrun:pipinstalldbt-snowflake-name:Déployerlesmodèles(dev)run:|
dbt deps
dbt seed --target dev
dbt run --target dev --full-refresh --select staging
dbt run --target dev --select marts+
dbt test --target dev
-name:Générerladocumentationrun:dbtdocsgenerate--targetdev-name:Uploadartifactsuses:actions/upload-pages-artifact@v3with:path:target/
7.2 Workflow de Déploiement
# Dev
dbt run --target dev
dbt test --target dev
# Staging (sur schéma dédié)
dbt run --target staging --select state:modified+ # Modèles modifiés uniquement
dbt test --target staging
# Production
dbt run --target prod --selector prod_selector
dbt test --target prod
dbt source freshness --target prod
8. Exposures (Traçabilité Dashboard)
# models/marts/reporting/exposures.ymlversion:2exposures:-name:oee_dashboardtype:dashboardmaturity:highurl:https://looker.company.com/dashboards/42description:"Tableau de bord OEE temps réel"depends_on:-ref('kpi_daily')-ref('fact_sensor_readings')owner:name:EVAemail:eva@thehive.local-name:machine_ml_featurestype:mlmaturity:mediumdescription:"Features ML pour la maintenance prédictive"depends_on:-ref('dim_machine')-ref('stg_sensors')owner:name:MLTeam
Pièges Courants (Pitfalls)
Modèles non idempotents.
Erreur : Un dbt run --full-refresh produit des résultats différents d'une exécution normale.
Correction : Toujours utiliser {{ ref() }} plutôt que les noms de tables bruts. Éviter les dépendances sur l'ordre d'exécution.
Modèles incrémentaux mal configurés.
Erreur :is_incremental() sans condition WHERE → rechargement complet à chaque run.
Correction : Toujours inclure un filtre WHERE timestamp > (SELECT MAX(timestamp) FROM {{ this }}).
Pas de documentation ni de tests.
Erreur : Un projet dbt sans schema.yml → impossible de comprendre la sémantique des colonnes.
Correction :schema.yml pour chaque modèle avec description et tests.
Tests trop longs (full scan).
Erreur : Exécuter tous les tests sur toute l'historique → coût élevé (BigQuery/ Snowflake).
Correction : Configurer severity: warn sur les tests coûteux, ou limiter aux dernières 24h.
Refs cassées après changement de nom de modèle.
Erreur : Renommer un modèle sans mettre à jour tous les {{ ref('ancien_nom') }}.
Correction : Utiliser dbt ls --resource-type model --output name pour auditer les dépendances. Configurer dbt build pour exécuter modèles dépendants en cascade.
Liste de Vérification (Checklist)
sources.yml définit toutes les données brutes avec freshness.
Chaque modèle a un schema.yml avec description et tests.
Les modèles incrémentaux ont la clause {% if is_incremental() %}.
{{ ref() }} utilisé partout (pas de nom de table en dur).
Snapshots configurés pour les dimensions à historique.
Tests unique + not_null sur toutes les surrogate keys.
Documentation générée (dbt docs generate).
CI/CD avec tests avant merge sur la branche principale.
Exposures définies pour les dashboards et artefacts ML.
