Journal d’exécution dbt : macros complètes
Résumé
Un hook
on-run-endenregistre dans une table, à chaque invocation de dbt, une ligne par nœud exécuté : statut, durée, lignes affectées, erreur, coût (BigQuery) et statut des dépendances. Aucun collecteur ni job externe : la capture est synchrone, au sein du run. Cette note donne le code complet d’un projet qui l’exécute : le modèle de la table, le hook et les huit macros. Le principe général est dans Observabilité des données.
Prérequis
dbt — Fondamentaux (modèles, macros, hooks), Observabilité des données pour le contexte, et des notions de Jinja.
Statut du code
Ce code provient d’un projet réel où il fonctionne. Il n’a pas été relancé de façon indépendante pour la rédaction de cette note, et il est testé sur BigQuery, DuckDB et Trino d’après ses commentaires : sur un autre moteur, ajouter la branche manquante de
upstream_ids_empty_cast. Les noms de variables sont à adapter au projet utilisé.
Le flux
flowchart LR R["dbt run, build ou test"] --> RR["objet results"] --> H["hook on-run-end"] --> L["log_dbt_results"] --> P["parse_dbt_results"] --> T[("table dbt_results")]
La clé de chaque ligne est invocation_id plus l’identifiant du nœud ; le champ result_id les concatène et sert de unique_key au modèle incrémental.
La configuration
# dbt_project.yml (extrait)
vars:
# Rétention en jours des lignes du journal
observability_retention_days: 90
# Noms du dataset et de la table, lus par la macro d'insertion : à changer ensemble
observability_dataset: observability
observability_dbt_results_table: dbt_results
models:
mon_projet:
00_observability:
+schema: "{{ var('observability_dataset', 'observability') }}"
on-run-end:
- "{{ log_dbt_results(results) }}"La table : un modèle qui crée la structure
Le modèle ne contient aucune donnée : il définit les colonnes avec un select filtré par where 1 = 0, et la macro d’insertion remplit la table. La rétention supprime en pre_hook les lignes dont la date de fin est plus ancienne que le seuil.
{{
config(
materialized = 'incremental',
transient = False,
unique_key = 'result_id',
on_schema_change = 'append_new_columns',
alias = var('observability_dbt_results_table'),
pre_hook = "{% if is_incremental() %} delete from {{ this }} where coalesce(completed_at, started_at) < '{{ observability_retention_cutoff() }}' {% endif %}"
)
}}
with empty_table as (
select
-- Key(s)
cast(null as {{ dbt.type_string() }}) as result_id,
cast(null as {{ dbt.type_string() }}) as invocation_id,
cast(null as {{ dbt.type_string() }}) as unique_id,
-- Temporal key(s) -- the "execute" phase of run_result.timing;
-- NULL for a skipped node (no "execute" phase)
cast(null as {{ dbt.type_string() }}) as started_at,
cast(null as {{ dbt.type_string() }}) as completed_at,
-- Dimensional / descriptive attributes
cast(null as {{ dbt.type_string() }}) as resource_type,
cast(null as {{ dbt.type_string() }}) as materialization,
-- flags.FULL_REFRESH for the whole invocation, not per-node --
-- lets a big rows_affected/bytes_billed spike be attributed to a
-- --full-refresh instead of misread as incremental drift.
cast(null as {{ dbt.type_boolean() }}) as is_full_refresh,
-- dbt's `group` config (node.group in the manifest); NULL for
-- ungrouped nodes -- most resource types don't have one. Named
-- group_name, not group, since GROUP is a SQL keyword.
cast(null as {{ dbt.type_string() }}) as group_name,
-- Generic test name (dbt/dbt_utils/custom), from test_metadata.
cast(null as {{ dbt.type_string() }}) as test_generic_name,
-- Package defining the test macro (dbt, dbt_utils, or this
-- project), from depends_on.macros.
cast(null as {{ dbt.type_string() }}) as test_package,
-- The specific column a test targets (node.column_name); more
-- reliable than parsing it out of db_fqn/unique_id.
cast(null as {{ dbt.type_string() }}) as test_column_name,
cast(null as {{ dbt.type_string() }}) as db_system,
cast(null as {{ dbt.type_string() }}) as db_name,
cast(null as {{ dbt.type_string() }}) as db_schema,
cast(null as {{ dbt.type_string() }}) as db_sql_table,
cast(null as {{ dbt.type_string() }}) as db_fqn,
-- Status
cast(null as {{ dbt.type_string() }}) as status,
-- Best-effort scrape of run_result.message's `type=`/`name=`
-- tokens (dbt's own exception message shape) -- not a structured
-- field dbt exposes, so empty on messages that don't contain them
-- (e.g. a plain test failure).
cast(null as {{ dbt.type_string() }}) as error_type,
cast(null as {{ dbt.type_string() }}) as error_name,
cast(null as {{ dbt.type_string() }}) as error_message,
-- Metrics / measures
cast(null as {{ dbt.type_float() }}) as execution_time_seconds,
cast(null as {{ dbt.type_int() }}) as rows_affected,
-- BigQuery-only (BigQueryAdapterResponse); always NULL on Trino
-- (TrinoAdapterResponse has no bytes_* fields at all) and DuckDB
-- (doesn't extend AdapterResponse beyond the base dbt-core class).
cast(null as {{ dbt.type_int() }}) as bytes_processed,
cast(null as {{ dbt.type_int() }}) as bytes_billed,
-- External system id
cast(null as {{ dbt.type_string() }}) as query_id,
-- Upstream aggregates -- did this node's dependencies succeed?
-- Source dependencies are excluded (they never appear in `results`,
-- so they'd always show as unknown and drown out the real signal).
{{ upstream_ids_empty_cast() }} as upstream_ids,
cast(null as {{ dbt.type_int() }}) as upstream_count,
cast(null as {{ dbt.type_boolean() }}) as upstream_any_failure,
cast(null as {{ dbt.type_int() }}) as upstream_success_count,
cast(null as {{ dbt.type_int() }}) as upstream_fail_count,
cast(null as {{ dbt.type_int() }}) as upstream_unknown_count,
-- Large payload, last
cast(null as {{ dbt.type_string() }}) as compiled_code
)
select * from empty_table
-- This is a filter so we will never actually insert these values
where 1 = 0Les colonnes sont typées avec les macros inter-bases de dbt (dbt.type_string() et suivantes), pour que le même modèle fonctionne sur plusieurs moteurs.
Les macros
log_dbt_results : le point d’entrée, appelé par le hook
Fichier obs/dbt_run/log_dbt_results.sql :
{% macro log_dbt_results(results) %}
{% set parsed_results = parse_dbt_results(results) %}
{% if parsed_results | length == 0 %}
{{ return('') }}
{% endif %}
{% set relation = target.database ~ '.' ~ var('observability_dataset') ~ '.' ~ var('observability_dbt_results_table') %}
{% set insert_columns = [
'result_id', 'invocation_id', 'unique_id', 'started_at', 'completed_at',
'resource_type', 'materialization', 'is_full_refresh', 'group_name', 'test_generic_name',
'test_package', 'test_column_name', 'db_system', 'db_name',
'db_schema', 'db_sql_table', 'db_fqn', 'status', 'error_type', 'error_name',
'error_message', 'execution_time_seconds', 'rows_affected',
'bytes_processed', 'bytes_billed', 'query_id', 'upstream_ids',
'upstream_count', 'upstream_any_failure', 'upstream_success_count',
'upstream_fail_count', 'upstream_unknown_count', 'compiled_code'
] %}
{% set value_rows = [] %}
{% for result in parsed_results %}
{% if result.upstream_ids | length > 0 %}
{% set upstream_ids_sql = upstream_ids_array_literal(result.upstream_ids) %}
{% else %}
{% set upstream_ids_sql = upstream_ids_empty_cast() %}
{% endif %}
{% set any_failure_sql %}{{ result.upstream_any_failure | upper }}{% endset %}
{% set column_values = {
'result_id': "'" ~ result.result_id ~ "'",
'invocation_id': "'" ~ result.invocation_id ~ "'",
'unique_id': "'" ~ result.unique_id ~ "'",
'started_at': sql_literal(result.started_at),
'completed_at': sql_literal(result.completed_at),
'resource_type': "'" ~ result.resource_type ~ "'",
'materialization': sql_literal(result.materialization),
'is_full_refresh': sql_literal(result.is_full_refresh),
'group_name': sql_literal(result.group_name),
'test_generic_name': sql_literal(result.test_generic_name),
'test_package': sql_literal(result.test_package),
'test_column_name': sql_literal(result.test_column_name),
'db_system': "'" ~ target.type ~ "'",
'db_name': sql_literal(result.db_name),
'db_schema': "'" ~ result.db_schema ~ "'",
'db_sql_table': "'" ~ result.db_sql_table ~ "'",
'db_fqn': "'" ~ result.db_fqn ~ "'",
'status': "'" ~ result.status ~ "'",
'error_type': sql_literal(result.error_type),
'error_name': sql_literal(result.error_name),
'error_message': sql_literal(result.error_message),
'execution_time_seconds': result.execution_time_seconds,
'rows_affected': result.rows_affected,
'bytes_processed': sql_literal(result.bytes_processed),
'bytes_billed': sql_literal(result.bytes_billed),
'query_id': sql_literal(result.query_id),
'upstream_ids': upstream_ids_sql,
'upstream_count': result.upstream_count,
'upstream_any_failure': any_failure_sql,
'upstream_success_count': result.upstream_success_count,
'upstream_fail_count': result.upstream_fail_count,
'upstream_unknown_count': result.upstream_unknown_count,
'compiled_code': sql_literal(result.compiled_code)
} %}
{% set row_values = [] %}
{% for col in insert_columns %}
{% do row_values.append(column_values[col]) %}
{% endfor %}
{% set row_sql = '(' ~ (row_values | join(', ')) ~ ')' %}
{% do value_rows.append(row_sql) %}
{% endfor %}
{#- Named columns, not positional: on_schema_change=append_new_columns
always ALTER TABLE ADD COLUMN at the end of the physical table on
BigQuery, regardless of this list's order, so a bare `values (...)`
insert would misalign as soon as they diverge. -#}
{% set insert_query %}
insert into {{ relation }} (
{{ insert_columns | join(', ') }}
)
values {{ value_rows | join(',\n') }}
{% endset %}
{% do run_query(insert_query) %}
{{ return('') }}
{% endmacro %}Insère les lignes dans la table. Il construit la liste des colonnes, rend chaque valeur en littéral SQL, puis exécute lui-même l’insertion avec run_query. L’insertion se fait par colonnes nommées, pas par position : on_schema_change: append_new_columns ajoute les colonnes à la fin de la table physique, ce qui décalerait un values (...) positionnel.
parse_dbt_results : aplatir chaque résultat
Fichier obs/dbt_run/parse_dbt_results.sql :
{% macro parse_dbt_results(results) %}
-- Create a list of parsed results
{%- set parsed_results = [] %}
-- Index this invocation's results by unique_id, for upstream lookups
{% set results_by_id = parse_dbt_results_index(results) %}
-- Flatten results and add to list
{% for run_result in results %}
-- Convert the run result object to a simple dictionary
{% set run_result_dict = run_result.to_dict() %}
-- Get the underlying dbt graph node that was executed
{% set node = run_result_dict.get('node') %}
{% set adapter_response = run_result_dict.get('adapter_response', {}) %}
{% set rows_affected = adapter_response.get('rows_affected', 0) %}
{%- if not rows_affected -%}
{% set rows_affected = 0 %}
{%- endif -%}
{% set query_id = adapter_response.get('query_id') or adapter_response.get('job_id') %}
-- Timing for the "execute" phase; absent for a skipped node.
{% set timing = run_result_dict.get('timing', []) %}
{% set execute_phase = (timing | selectattr('name', 'equalto', 'execute') | list | first) %}
-- Best-effort scrape of dbt's own exception message shape:
-- "... type=<X>, name=<Y> ...". Not a structured field dbt
-- exposes -- empty on messages that don't contain those tokens
-- (e.g. a plain test failure with no exception).
{% set raw_message = run_result_dict.get('message', '') or '' %}
{% set error_type = '' %}
{% set error_name = '' %}
{% if raw_message %}
{% set type_start = raw_message.find('type=') %}
{% if type_start != -1 %}
{% set type_value = raw_message[type_start + 5:] %}
{% set type_end = type_value.find(',') %}
{% if type_end == -1 %}{% set type_end = type_value.find(')') %}{% endif %}
{% if type_end != -1 %}{% set error_type = type_value[:type_end] %}{% endif %}
{% endif %}
{% set name_start = raw_message.find('name=') %}
{% if name_start != -1 %}
{% set name_value = raw_message[name_start + 5:] %}
{% set name_end = name_value.find(',') %}
{% if name_end == -1 %}{% set name_end = name_value.find(')') %}{% endif %}
{% if name_end != -1 %}{% set error_name = name_value[:name_end] %}{% endif %}
{% endif %}
{% endif %}
{% set db_name = node.get('database') %}
{% set db_fqn_parts = [db_name, node.get('schema'), (node.get('alias') or node.get('name'))] %}
-- Upstream dependencies' statuses, for root-cause attribution.
{% set upstream_data = parse_dbt_results_upstream(node, results_by_id) %}
{% set upstream_aggregates = parse_dbt_results_aggregates(upstream_data.ids, upstream_data.statuses) %}
-- Lowercased for consistency: freshness check results
-- (resource_type='source', logged through this same on-run-end
-- hook -- dbt-fusion does fire it for `dbt source freshness`,
-- unlike classic dbt-core) come back as "Pass"/"Warn"/"Error",
-- capitalized, while model/test statuses are already lowercase.
{% set status_value = (run_result_dict.get('status') or '') | lower %}
{#- Real test name from the manifest, not a unique_id guess. -#}
{% set test_generic_name = node.get('test_metadata', {}).get('name') if node.get('test_metadata') else none %}
{#- Package owning the test macro, resolved from depends_on.macros
(macro.<package>.test_<name>) -- generic, no per-test list.
namespace() is required here, not a bare {% set %}: Jinja
for-loops create a new scope per iteration, so a plain
{% set test_package = ... %} inside the loop would never be
visible after it ends. -#}
{% set ns = namespace(test_package=none) %}
{% if test_generic_name %}
{% set expected_suffix = '.test_' ~ test_generic_name %}
{% for macro_id in node.get('depends_on', {}).get('macros', []) %}
{% if ns.test_package is none and macro_id.endswith(expected_suffix) %}
{% set ns.test_package = macro_id.split('.')[1] %}
{% endif %}
{% endfor %}
{% endif %}
{% set parsed_result_dict = {
'result_id': invocation_id ~ '.' ~ node.get('unique_id'),
'invocation_id': invocation_id,
'unique_id': node.get('unique_id'),
'started_at': execute_phase.get('started_at') if execute_phase else none,
'completed_at': execute_phase.get('completed_at') if execute_phase else none,
'resource_type': node.get('resource_type'),
'materialization': node.get('config', {}).get('materialized'),
'is_full_refresh': flags.FULL_REFRESH,
'group_name': node.get('config', {}).get('group') or node.get('group'),
'test_generic_name': test_generic_name,
'test_package': ns.test_package,
'test_column_name': node.get('column_name'),
'db_system': target.type,
'db_name': db_name,
'db_schema': node.get('schema'),
'db_sql_table': node.get('alias') or node.get('name'),
'db_fqn': db_fqn_parts | join('.'),
'status': status_value,
'error_type': error_type,
'error_name': error_name,
'error_message': raw_message,
'execution_time_seconds': run_result_dict.get('execution_time'),
'rows_affected': rows_affected,
'bytes_processed': adapter_response.get('bytes_processed'),
'bytes_billed': adapter_response.get('bytes_billed'),
'query_id': query_id,
'upstream_ids': upstream_data.ids,
'upstream_count': upstream_aggregates.upstream_count,
'upstream_any_failure': upstream_aggregates.upstream_any_failure,
'upstream_success_count': upstream_aggregates.upstream_success_count,
'upstream_fail_count': upstream_aggregates.upstream_fail_count,
'upstream_unknown_count': upstream_aggregates.upstream_unknown_count,
'compiled_code': node.get('compiled_code'),
}%}
{% do parsed_results.append(parsed_result_dict) %}
{% endfor %}
{{ return(parsed_results) }}
{% endmacro %}Transforme chaque résultat de run en dictionnaire : identifiants, temps, statut, lignes affectées, erreur, nom du test et son package, coût (BigQuery), dépendances. Le message d’erreur est analysé au mieux (type=, name=) : dbt n’expose pas ces champs sous forme structurée. namespace() est nécessaire pour garder une variable modifiée dans une boucle Jinja.
parse_dbt_results_index, parse_dbt_results_upstream, parse_dbt_results_aggregates : les dépendances
Fichier obs/dbt_run/_parse_dbt_results_index.sql :
{% macro parse_dbt_results_index(results) %}
{#- Builds a unique_id -> run_result lookup for this invocation, so
each node's upstream dependencies can be resolved by status
without an O(n) scan per node. -#}
{% set results_by_id = {} %}
{% for run_result in results %}
{% set rid = run_result.to_dict().get('node', {}).get('unique_id', '') %}
{% if rid %}
{% set _ = results_by_id.update({rid: run_result}) %}
{% endif %}
{% endfor %}
{{ return(results_by_id) }}
{% endmacro %}Fichier obs/dbt_run/_parse_dbt_results_upstream.sql :
{% macro parse_dbt_results_upstream(node, results_by_id) %}
{#- Source dependencies are excluded, always -- a source never appears
in `results` (dbt doesn't "run" a source), so counting it would
always contribute an unknown status and drown out the real signal
for models whose only upstream is a source. -#}
{% set all_upstreams = node.get('depends_on', {}).get('nodes', []) %}
{% set filtered = [] %}
{% for up_id in all_upstreams %}
{% if up_id[:7] != 'source.' %}{% do filtered.append(up_id) %}{% endif %}
{% endfor %}
{% set all_upstreams = filtered %}
{% set upstream_statuses = [] %}
{% for up_id in all_upstreams %}
{% set up_result = results_by_id.get(up_id) %}
{% if up_result %}
{% set _ = upstream_statuses.append(up_result.to_dict().get('status', 'unknown')) %}
{% else %}
{% set _ = upstream_statuses.append('unknown') %}
{% endif %}
{% endfor %}
{{ return({'ids': all_upstreams, 'statuses': upstream_statuses}) }}
{% endmacro %}Les dépendances de type source sont exclues : une source n’apparaît jamais dans les résultats, elle donnerait toujours un statut inconnu.
Fichier obs/dbt_run/_parse_dbt_results_aggregates.sql :
{% macro parse_dbt_results_aggregates(upstream_ids, upstream_statuses) %}
{#- A failed model reports status='error', a failed test reports
status='fail' (or 'error' on a runtime error) -- both count as a
failure here, since upstream nodes are commonly models. #}
{% set upstream_count = upstream_ids | length %}
{% set upstream_success_count = upstream_statuses | select('in', ['success', 'pass']) | list | length %}
{% set upstream_fail_count = upstream_statuses | select('in', ['fail', 'error']) | list | length %}
{% set upstream_unknown_count = upstream_statuses | select('equalto', 'unknown') | list | length %}
{% set upstream_any_failure = upstream_fail_count > 0 %}
{{ return({
'upstream_count': upstream_count,
'upstream_any_failure': upstream_any_failure,
'upstream_success_count': upstream_success_count,
'upstream_fail_count': upstream_fail_count,
'upstream_unknown_count': upstream_unknown_count
}) }}
{% endmacro %}Un modèle en échec porte le statut error, un test en échec fail (ou error sur une erreur d’exécution) : les deux comptent comme un échec en amont.
upstream_ids_empty_cast et upstream_ids_array_literal : les tableaux d’identifiants
Fichier obs/dbt_run/_upstream_ids_empty.sql :
{% macro upstream_ids_empty_cast() %}
{#- "No upstreams" cast for `upstream_ids`, one branch per verified
engine. No catch-all else branch -- an unverified engine should
fail loudly, not guess wrong syntax. -#}
{%- if target.type == 'bigquery' -%}
CAST(NULL AS ARRAY<STRING>)
{%- elif target.type == 'trino' -%}
CAST(NULL AS ARRAY(VARCHAR(256)))
{%- elif target.type == 'duckdb' -%}
CAST(NULL AS VARCHAR[])
{%- else -%}
{{ exceptions.raise_compiler_error(
"upstream_ids_empty_cast: no verified ARRAY-of-string cast for target.type='"
~ target.type ~ "'. Add a branch here once you've confirmed the correct syntax "
~ "(and nullability!) for this engine -- don't assume it matches another engine's."
) }}
{%- endif -%}
{% endmacro %}
{% macro upstream_ids_array_literal(ids) %}
{#- Non-empty array literal: ARRAY[...] works on BigQuery/Trino. -#}
ARRAY[{% for id in ids %}{{ sql_literal(id) }}{% if not loop.last %}, {% endif %}{% endfor %}]
{% endmacro %}Un cast par moteur vérifié, et volontairement aucune branche par défaut : un moteur non vérifié doit échouer clairement plutôt que deviner une syntaxe.
sql_literal : rendre une valeur en littéral SQL
Fichier obs/sql/sql_literal.sql :
{#- Renders a Python value as a SQL literal (NULL-safe) for use inside
hand-built INSERT statements. Dispatches on the Jinja value's type:
- none -> NULL
- boolean -> bare TRUE/FALSE (valid on BigQuery, DuckDB, and Trino alike)
- number -> bare numeric literal
- anything else -> quoted string literal via dbt's adapter-dispatched
cross-database macros (https://docs.getdbt.com/reference/dbt-jinja-functions/cross-database-macros),
which already handle BigQuery vs. ANSI quote/newline escaping.
Why numbers/booleans must stay bare (not quoted) -- verified against a
live Trino 482 container, not just docs:
insert into t (n integer) values ('42') -> fails: "Insert query
has mismatched column types: Table: [integer], Query: [varchar(2)]"
insert into t (b boolean) values ('true') -> same mismatch error
Trino (like BigQuery, per its docs: "String literals don't coerce to
numeric types") does NOT implicitly coerce a quoted string into a
numeric/boolean column on INSERT. DuckDB is the permissive outlier and
tolerates the quoted form, so bare rendering is required for portability,
not optional.
Why default__escape_single_quotes (doubled '' ) is correct for Trino:
verified on the same container that dbt's ANSI-default escaping round-trips
correctly, e.g. escape_single_quotes("they're ain't a word") ->
"they''re ain''t a word", inserted and read back as "they're ain't a word".
BigQuery also treats backslash as an escape character inside string
literals (unlike DuckDB/Trino's ANSI-default literals), so a value
containing a literal backslash not already valid as a BigQuery escape
(e.g. compiled_code containing "\." from a regex) raises "Illegal escape
sequence" once inlined. Backslashes must be doubled before quoting, and
only for BigQuery -- doubling them for DuckDB/Trino would corrupt the
round-trip since those dialects don't unescape backslashes at all. -#}
{% macro sql_literal(value) %}
{%- if value is none -%}
NULL
{%- elif value is boolean -%}
{{ value | upper }}
{%- elif value is number -%}
{{ value }}
{%- else -%}
{%- set string_value = value | string -%}
{%- if target.type == 'bigquery' -%}
{%- set string_value = string_value | replace('\\', '\\\\') -%}
{%- endif -%}
{{ dbt.string_literal(dbt.escape_single_quotes(string_value)) }}
{%- endif -%}
{% endmacro %}Gère NULL, les booléens, les nombres (laissés nus, car BigQuery et Trino ne convertissent pas une chaîne en nombre à l’insertion) et les chaînes, avec un traitement des barres obliques inverses propre à BigQuery.
observability_retention_cutoff : la date limite de rétention
Fichier obs/dbt_run/_observability_retention_cutoff.sql :
{% macro observability_retention_cutoff() %}
{#- ISO cutoff string, formatted like started_at/completed_at/valid_to
are actually stored, so a plain VARCHAR `<` works on every engine
with no per-adapter date function needed. -#}
{%- set retention_days = var('observability_retention_days') -%}
{%- set cutoff = run_started_at - modules.datetime.timedelta(days=retention_days) -%}
{{ return(cutoff.strftime('%Y-%m-%dT%H:%M:%S.%fZ')) }}
{% endmacro %}Renvoie une chaîne ISO au même format que les horodatages stockés, ce qui permet une comparaison < sur du texte, sans fonction de date propre à un moteur.
Limites connues
- Un seul moteur de référence par type de cast : les tableaux d’identifiants demandent une branche par moteur.
- Analyse du message d’erreur :
type=etname=sont extraits d’un format de message non garanti par dbt. - Insertion en une requête : un très grand nombre de nœuds produit un
insertvolumineux. - Pas de fraîcheur des sources, de distribution ni de schéma : ce journal couvre le volume et l’exécution ; voir Observabilité des données.
Notes liées
- Observabilité des données : le principe, les six angles, les requêtes de contrôle.
- dbt — Fondamentaux : hooks, macros et matérialisations.
- dbt_bigquery_monitoring : le coût côté BigQuery, par un package.
- Idempotence et backfill :
unique_keyet rejeu. - Glossaire data engineering : lineage.
- MOC Data Engineering : carte du vault.