Créer une matérialisation personnalisée dans dbt

Parfois, la matérialisation table standard n’est pas assez fiable. Vous reconstruisez une table critique et une requête de dashboard tombe pile entre le DROP et le RENAME. Ou un mauvais filtre vide votre SQL, et dbt remplace une table d’un million de lignes par zéro lignes. Ou encore, BigQuery supprime les row access policies que vous avez passé un après-midi à configurer parce que dbt a recréé la table de zéro.

C’est là qu’on construit la sienne.

Une matérialisation personnalisée, ça peut paraître intimidant, mais ce n’est qu’une macro avec une structure bien précise. Une fois l’anatomie comprise, vous pouvez créer des matérialisations qui correspondent exactement à la façon dont votre warehouse doit construire ses tables.

Ce tutoriel vous guide dans la construction de deux matérialisations concrètes : un swap de table sans interruption avec validation, et une table sécurisée qui réapplique automatiquement les politiques de sécurité après chaque build. Les deux résolvent des problèmes que les matérialisations natives de dbt ne couvrent pas.

Ce que dbt gère déjà

Avant d’écrire une matérialisation personnalisée, vérifiez si les options natives résolvent votre problème. La matérialisation incrémentale de dbt est plus flexible qu’il n’y paraît.

Le comportement append-only est natif. Configurez incremental_strategy='append' pour insérer des lignes sans merge ni mise à jour, et ajoutez full_refresh: false pour empêcher --full-refresh de supprimer la table.

{{ config(
    materialized='incremental',
    incremental_strategy='append',
    full_refresh=false
) }}

Les overwrites au niveau des partitions sont également natifs. La stratégie insert_overwrite de dbt-bigquery remplace des partitions entières au lieu de merger des lignes individuelles. Elle construit une table temporaire, identifie les partitions concernées, et les swap.

{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={'field': 'event_date', 'data_type': 'date'}
) }}

La stratégie delete+insert gère le remplacement ligne par ligne via une clé unique. Entre ces trois stratégies et la config full_refresh, la plupart des patterns incrémentaux sont couverts sans code personnalisé.

Les matérialisations personnalisées servent à combler les lacunes restantes : le contrôle du processus de swap lui-même, la validation avant remplacement, ou les opérations post-build qui dépassent ce que les hooks gèrent proprement.

Anatomie d’une matérialisation

Chaque matérialisation suit la même structure en six étapes :

{%- materialization my_custom_mat, adapter='default' -%}

  {# 1. SETUP - Préparer les relations #}
  {% set target_relation = this.incorporate(type='table') %}
  {% set existing_relation = load_cached_relation(this) %}

  {# 2. PRE-HOOKS #}
  {{ run_hooks(pre_hooks) }}

  {# 3. SQL PRINCIPAL - Construire la relation #}
  {% call statement('main') %}
    {{ sql }}
  {% endcall %}

  {# 4. POST-HOOKS #}
  {{ run_hooks(post_hooks) }}

  {# 5. NETTOYAGE - Grants, docs, permissions #}
  {% do apply_grants(target_relation, grant_config, should_revoke) %}
  {% do persist_docs(target_relation, model) %}

  {# 6. COMMIT ET RETOUR #}
  {% do adapter.commit() %}
  {{ return({'relations': [target_relation]}) }}

{%- endmaterialization -%}

Les éléments clés :

• this est la relation cible (database.schema.model_name)
• sql contient la requête SELECT compilée du fichier modèle
• config contient la configuration du modèle (tout ce qui est dans le bloc config())
• statement('main') exécute le SQL et le rend visible dans les logs
• adapter fournit les méthodes spécifiques à la base de données

Vous utiliserez régulièrement quelques méthodes de l’adapter :

{# Vérifier si la relation existe déjà #}
{% set existing = load_cached_relation(this) %}

{# Obtenir les infos de colonnes d'une relation #}
{% set columns = adapter.get_columns_in_relation(target_relation) %}

{# Supprimer une relation #}
{% do adapter.drop_relation(old_relation) %}

{# Renommer une relation #}
{% do adapter.rename_relation(temp_relation, target_relation) %}

La fonction load_cached_relation() retourne la relation si elle existe, ou none sinon. C’est ainsi que vous distinguez la première exécution des suivantes.

Construire une matérialisation table sans interruption

La matérialisation table standard de dbt construit une table temporaire, supprime l’existante, puis renomme la temporaire à sa place. Ça fonctionne pour la plupart des modèles. Ça pose problème pour les tables critiques en production de deux façons.

Il y a une brève fenêtre entre le DROP et le RENAME pendant laquelle la table n’existe pas. Si une requête de dashboard ou un extract planifié s’exécute pendant cette fenêtre, il échoue avec une erreur « table not found ». Pour les modèles internes, personne ne s’en aperçoit. Pour la table qui alimente un dashboard client, c’est un ticket de support.

Il n’y a pas non plus d’étape de validation. Si votre SQL retourne zéro ligne parce qu’une table source est vide ou qu’un filtre de date est incorrect, la matérialisation standard remplace l’ancienne table par une table vide. Vous ne le découvrez que quand quelqu’un remarque des données manquantes.

Une matérialisation sans interruption résout les deux problèmes : elle construit sous un nom temporaire, valide le résultat, puis swap via un rename. L’ancienne table reste interrogeable tant que la nouvelle n’est pas confirmée.

Créez macros/materializations/zero_downtime_table.sql :

{%- materialization zero_downtime_table, adapter='bigquery' -%}

  {# Configuration #}
  {% set min_row_count = config.get('min_row_count', 1) %}

  {# Préparer les relations #}
  {% set target_relation = this.incorporate(type='table') %}
  {% set existing_relation = load_cached_relation(this) %}
  {% set temp_relation = make_temp_relation(target_relation) %}
  {% set backup_relation = make_backup_relation(target_relation) %}

  {# Gérer l'incompatibilité de type (l'existant est une vue, pas une table) #}
  {% if existing_relation is not none and existing_relation.type != 'table' %}
    {% do adapter.drop_relation(existing_relation) %}
    {% set existing_relation = none %}
  {% endif %}

  {{ run_hooks(pre_hooks) }}

  {# Construire la nouvelle table sous un nom temporaire. L'ancienne reste interrogeable. #}
  {% call statement('main') %}
    {{ create_table_as(false, temp_relation, sql) }}
  {% endcall %}

  {# Valider avant le swap #}
  {% call statement('validate', fetch_result=True) %}
    SELECT COUNT(*) AS row_count FROM {{ temp_relation }}
  {% endcall %}

  {% set row_count = load_result('validate')['data'][0][0] %}

  {% if row_count < min_row_count %}
    {% do adapter.drop_relation(temp_relation) %}
    {{ exceptions.raise_compiler_error(
      "Validation failed: " ~ row_count ~ " rows, expected at least "
      ~ min_row_count ~ ". Old table preserved."
    ) }}
  {% endif %}

  {# Swap : renommer l'existante en backup, puis la temporaire en cible #}
  {% if existing_relation is not none %}
    {% do adapter.rename_relation(existing_relation, backup_relation) %}
  {% endif %}

  {% do adapter.rename_relation(temp_relation, target_relation) %}

  {# Supprimer le backup après un swap réussi #}
  {% if existing_relation is not none %}
    {% do adapter.drop_relation(backup_relation) %}
  {% endif %}

  {{ run_hooks(post_hooks) }}

  {% set grant_config = config.get('grants') %}
  {% do apply_grants(target_relation, grant_config) %}
  {% do persist_docs(target_relation, model) %}

  {% do adapter.commit() %}
  {{ return({'relations': [target_relation]}) }}

{%- endmaterialization -%}

L’ordre des opérations est important. L’ancienne table est renommée en backup avant que la nouvelle prenne sa place. Si le build ou la validation échoue, l’ancienne table reste intacte. Le backup n’est supprimé qu’après la réussite du swap.

La fenêtre pendant laquelle ni l’ancienne ni la nouvelle table n’existe sous le nom cible correspond au temps de deux opérations de métadonnées BigQuery (renommer l’ancienne en backup, renommer la temporaire en cible). C’est typiquement moins d’une seconde, contre une interruption potentielle de plusieurs minutes lors d’un rebuild standard. BigQuery ne supporte pas les swaps atomiques comme le ALTER TABLE ... SWAP WITH de Snowflake, donc cette fenêtre ne peut pas être totalement éliminée, mais elle est assez courte pour que la plupart des requêtes concurrentes ne soient pas affectées.

Utilisez-le dans un modèle :

{{ config(
    materialized='zero_downtime_table',
    min_row_count=1000
) }}

SELECT
    user_id,
    email,
    account_status,
    created_at
FROM {{ ref('int__users_joined') }}
WHERE account_status != 'deleted'

Si le modèle retourne moins de 1 000 lignes, dbt lève une erreur et l’ancienne table reste en place. Les logs vous disent exactement ce qui s’est passé.

Comparer avec le build précédent

Un seuil statique de nombre de lignes détecte les tables vides, mais une vérification relative détecte des problèmes plus subtils. Vous pouvez comparer la nouvelle table avec l’existante pour signaler des baisses de volume inattendues :

{# Ajouter après la vérification de min_row_count #}
{% set max_drop_pct = config.get('max_row_count_drop_pct', 50) %}

{% if existing_relation is not none and max_drop_pct is not none %}
  {% call statement('old_count', fetch_result=True) %}
    SELECT COUNT(*) FROM {{ existing_relation }}
  {% endcall %}
  {% set old_count = load_result('old_count')['data'][0][0] %}

  {% if old_count > 0 %}
    {% set drop_pct = ((old_count - row_count) * 100 / old_count) | int %}
    {% if drop_pct > max_drop_pct %}
      {% do adapter.drop_relation(temp_relation) %}
      {{ exceptions.raise_compiler_error(
        "Row count dropped " ~ drop_pct ~ "% (" ~ old_count ~ " to " ~ row_count
        ~ "), threshold is " ~ max_drop_pct ~ "%. Old table preserved."
      ) }}
    {% endif %}
  {% endif %}
{% endif %}

Cela couvre le scénario où une table source se vide partiellement ou un filtre supprime trop de données. Le modèle échoue de manière sûre plutôt que de propager des données incomplètes en aval.

Pour les tables dont le nombre de lignes varie naturellement, configurez max_row_count_drop_pct à none dans la config du modèle pour désactiver la vérification relative tout en conservant le minimum absolu.

Construire une matérialisation table sécurisée

BigQuery supporte les row access policies qui restreignent les lignes visibles par chaque groupe d’utilisateurs. Quand dbt supprime et recrée une table (comme le fait la matérialisation table), toutes les row access policies de cette table disparaissent. Il faut les réappliquer après chaque build.

Les post-hooks peuvent le faire. Pour une ou deux policies, ça suffit. Mais une table avec trois row access policies et huit descriptions de colonnes, c’est une dizaine de requêtes SQL dans les post-hooks. C’est difficile à maintenir et facile à rater, surtout quand le même pattern de sécurité s’applique sur trente modèles.

Une matérialisation personnalisée rend tout ça déclaratif. Vous définissez les policies dans la config, et la matérialisation les applique automatiquement après la construction de la table.

Créez macros/materializations/secured_table.sql :

{%- materialization secured_table, adapter='bigquery' -%}

  {# Préparer les relations #}
  {% set target_relation = this.incorporate(type='table') %}
  {% set existing_relation = load_cached_relation(this) %}

  {{ run_hooks(pre_hooks) }}

  {# Supprimer et reconstruire #}
  {% if existing_relation is not none %}
    {% do adapter.drop_relation(existing_relation) %}
  {% endif %}

  {% call statement('main') %}
    {{ create_table_as(false, target_relation, sql) }}
  {% endcall %}

  {# Appliquer les descriptions de colonnes depuis la config #}
  {% set column_descriptions = config.get('column_descriptions', {}) %}
  {% for col_name, description in column_descriptions.items() %}
    {% call statement('describe_' ~ col_name) %}
      ALTER TABLE {{ target_relation }}
      ALTER COLUMN {{ col_name }}
      SET OPTIONS(description='{{ description | replace("'", "\\'") }}')
    {% endcall %}
  {% endfor %}

  {# Appliquer les row access policies depuis la config #}
  {% set row_access_policies = config.get('row_access_policies', []) %}
  {% for policy in row_access_policies %}
    {% call statement('policy_' ~ policy.name) %}
      CREATE OR REPLACE ROW ACCESS POLICY {{ policy.name }}
      ON {{ target_relation }}
      GRANT TO ({{ policy.grantees | join(', ') }})
      FILTER USING ({{ policy.filter_expression }})
    {% endcall %}
  {% endfor %}

  {{ run_hooks(post_hooks) }}

  {% set grant_config = config.get('grants') %}
  {% do apply_grants(target_relation, grant_config) %}
  {% do persist_docs(target_relation, model) %}

  {% do adapter.commit() %}
  {{ return({'relations': [target_relation]}) }}

{%- endmaterialization -%}

Utilisez-le dans un modèle :

{{ config(
    materialized='secured_table',
    column_descriptions={
        'user_id': 'Unique identifier from the auth system',
        'email': 'User email address, restricted to PII-authorized groups',
        'region': 'Geographic region code used for access filtering',
        'lifetime_value': 'Total revenue attributed to this user in EUR'
    },
    row_access_policies=[
        {
            'name': 'eu_analysts_filter',
            'grantees': ['"group:eu-analysts@company.com"'],
            'filter_expression': "region = 'EU'"
        },
        {
            'name': 'us_analysts_filter',
            'grantees': ['"group:us-analysts@company.com"'],
            'filter_expression': "region = 'US'"
        },
        {
            'name': 'admin_full_access',
            'grantees': ['"group:data-admins@company.com"'],
            'filter_expression': 'TRUE'
        }
    ]
) }}

SELECT
    user_id,
    email,
    region,
    lifetime_value
FROM {{ ref('int__users_enriched') }}

À chaque exécution, le modèle est reconstruit avec les bonnes descriptions de colonnes et les bonnes row access policies. Les analystes EU ne voient que les lignes EU, les analystes US que les lignes US, et les admins data voient tout. Pas d’étapes manuelles, pas de post-hooks oubliés.

Pourquoi pas les post-hooks ?

Pour un seul modèle avec une policy, les post-hooks fonctionnent :

{{ config(
    materialized='table',
    post_hook="CREATE OR REPLACE ROW ACCESS POLICY region_filter ON {{ this }} GRANT TO (\"group:eu-analysts@company.com\") FILTER USING (region = 'EU')"
) }}

Ça cesse de passer à l’échelle à partir de trois policies par modèle. Le bloc config devient illisible, les guillemets deviennent fragiles (doubles guillemets imbriqués dans des chaînes SQL dans du Jinja), et appliquer le même pattern sur plusieurs modèles revient à copier des murs de texte post-hook.

Une matérialisation prend un dict de config et génère le DDL approprié. Ajouter une nouvelle policy, c’est une entrée de dict, pas une chaîne SQL soigneusement échappée.

Les descriptions de colonnes ont une alternative : le persist_docs de dbt avec columns: true lit les descriptions depuis les fichiers schema YAML et les applique via l’adapter. Ça fonctionne bien si votre équipe maintient déjà les fichiers schema YAML. L’approche par matérialisation garde les descriptions dans la config du modèle à côté du SQL, ce que certaines équipes préfèrent. Les deux approches sont valides.

Étendre avec le data masking

La sécurité au niveau des colonnes dans BigQuery utilise les policy tags de Data Catalog. Le pattern est similaire aux row access policies, mais le DDL assigne des policy tags aux colonnes plutôt que de créer des policies autonomes :

{# Ajouter ce bloc après les row access policies #}
{% set column_policy_tags = config.get('column_policy_tags', {}) %}
{% for col_name, tag_path in column_policy_tags.items() %}
  {% call statement('tag_' ~ col_name) %}
    ALTER TABLE {{ target_relation }}
    ALTER COLUMN {{ col_name }}
    SET OPTIONS(policy_tags=['{{ tag_path }}'])
  {% endcall %}
{% endfor %}

Puis dans la config du modèle :

{{ config(
    materialized='secured_table',
    column_policy_tags={
        'email': 'projects/my-project/locations/us/taxonomies/123/policyTags/456',
        'phone': 'projects/my-project/locations/us/taxonomies/123/policyTags/789'
    }
) }}

La taxonomie et les policy tags doivent exister dans Data Catalog avant que la matérialisation ne les référence. Créez-les séparément dans le cadre de votre setup d’infrastructure. La matérialisation gère uniquement l’assignation au niveau des colonnes à chaque rebuild.

Déboguer et tester les matérialisations

Les matérialisations personnalisées échouent de manière prévisible. Voici les erreurs que vous rencontrerez et leurs causes.

« Relation does not exist » signifie généralement que vous référencez une relation absente du cache. Utilisez load_cached_relation() au lieu de requêter directement la base de données, ou appelez adapter.cache.clear() si vous avez modifié des relations en dehors du cache.

« Transaction rolled back » ou des changements qui ne persistent pas signifie que vous avez oublié adapter.commit(). Appelez-le toujours avant le return.

« Relation type mismatch » survient quand la relation existante est une vue mais que vous essayez de la manipuler comme une table. Utilisez this.incorporate(type='table') et gérez les incompatibilités de type explicitement :

{% if existing_relation is not none and existing_relation.type != 'table' %}
  {% do adapter.drop_relation(existing_relation) %}
  {% set existing_relation = none %}
{% endif %}

Tester une nouvelle matérialisation

Créez un modèle de test simple qui utilise votre matérialisation :

-- models/test_zero_downtime.sql
{{ config(
    materialized='zero_downtime_table',
    min_row_count=1
) }}

SELECT 1 AS id, 'test' AS value

Exécutez-le deux fois :

dbt run --select test_zero_downtime
dbt run --select test_zero_downtime

Après les deux exécutions, requêtez la table cible. Vous devriez voir une ligne reconstruite proprement via le swap.

Consultez target/compiled/ pour voir le SQL réellement généré par votre matérialisation. C’est précieux pour le débogage, car vous verrez exactement quelle requête dbt a tenté d’exécuter.

Pour la table sécurisée, vérifiez que les row access policies ont bien été appliquées :

SELECT *
FROM `region-us`.INFORMATION_SCHEMA.OBJECT_PRIVILEGES
WHERE object_name = 'your_model_name'

Quand écrire une matérialisation personnalisée (et quand s’en abstenir)

Les matérialisations personnalisées ajoutent de la charge de maintenance. Considérez d’abord les alternatives.

Les post-hooks couvrent beaucoup de cas. Si vous avez besoin d’un seul GRANT après le build ou d’un ALTER TABLE, un hook est plus simple qu’une matérialisation. Les macros dans le SQL du modèle peuvent aussi générer des requêtes complexes sans toucher aux matérialisations.

Écrivez une matérialisation personnalisée quand le pattern est suffisamment complexe pour que le copier-coller soit source d’erreurs, quand vous avez besoin du comportement sur de nombreux modèles, ou quand les matérialisations natives vont à l’encontre de ce que vous cherchez. Le pattern sans interruption est pertinent quand vos tables de production ne tolèrent pas d’interruption pendant le build ni de swaps non validés. Le pattern table sécurisée est pertinent quand la sécurité au niveau des lignes s’applique sur des dizaines de modèles et que vous voulez un seul endroit pour maintenir la logique des policies.

Passez-vous de la matérialisation personnalisée quand un post-hook résout le problème, quand seulement un ou deux modèles ont besoin du comportement, ou quand vous êtes la seule personne à le maintenir.

Le pattern sans interruption est le plus simple des deux et un bon point de départ pour apprendre la structure. Le pattern table sécurisée suit les mêmes principes avec plus de composants. Les deux patterns résolvent des problèmes récurrents dès que votre projet dbt passe du développement à la production, là où les consommateurs en aval et la gouvernance des données comptent vraiment.