Quali sono i principi su cui si basa un ideale Data Warehouse?

Focus sul valore per il business e sull'analitica, senza codice boilerplate. Gestione del DWH come un codice sorgente: versioning, revisione, test automatici e CI. Modularità, scalabilità, codice aperto e comunità. Documentazione utente amichevole e visualizzazione delle dipendenze (Data Lineage).

Per tutto questo e per il ruolo di DBT nell'ecosistema Big Data & Analytics — benvenuti sotto il cat.

Ciao a tutti

Sono Artemij Kozyr. Da oltre 5 anni lavoro con i data warehouse, occupandomi di costruzione di ETL/ELT, nonché analisi dei dati e visualizzazione. Attualmente lavoro in Wheely, insegno in OTUS nel corso di Data Engineer, e oggi voglio condividere con voi un articolo che ho scritto in vista dell'inizio di un nuovo gruppo nel corso.

Panoramica

Il framework DBT è tutto incentrato sulla lettera T nell'acronimo ELT (Extract — Transform — Load).

Con l'emergere di database analitici così performanti e scalabili come BigQuery, Redshift e Snowflake, non ha più senso effettuare trasformazioni al di fuori del Data Warehouse.

DBT non estrae dati dalle sorgenti, ma offre enormi opportunità per lavorare con i dati già caricati nello Storage (nel Internal o External Storage).

Lo scopo principale di DBT è prendere il codice, compilarlo in SQL, eseguire i comandi nell'ordine corretto nello Storage.

Struttura del progetto DBT

Il progetto è composto da directory e file di soli 2 tipi:

Modello (.sql) — unità di trasformazione, espressa tramite una query SELECT
File di configurazione (.yml) — parametri, impostazioni, test, documentazione

A livello base, il lavoro è strutturato come segue:

L'utente prepara il codice dei modelli in qualsiasi IDE comoda
Usando la CLI, viene chiamato l'avvio dei modelli, DBT compila il codice dei modelli in SQL
Il codice SQL compilato viene eseguito nello Storage nell'ordine specificato (grafo)

Ecco come può apparire l'avvio dalla CLI:

Tutto è SELECT

Questa è la killer feature del framework Data Build Tool. In altre parole, DBT astrae tutto il codice relativo alla materializzazione delle tue query nello Storage (variazioni dei comandi CREATE, INSERT, UPDATE, DELETE, ALTER, GRANT, …).

Qualsiasi modello implica la scrittura di una singola query SELECT, che definisce il set di dati risultante.

La logica delle trasformazioni può essere multilivello e consolidare dati provenienti da diversi altri modelli. Ecco un esempio di un modello che costruirà una vetrina degli ordini (f_orders):

{% set payment_methods = ['credit_card', 'coupon', 'bank_transfer', 'gift_card'] %}
 
con ordini come (
 
   select * from {{ ref('stg_orders') }}
 
),
 
pagamenti_ordine come (
 
   select * from {{ ref('order_payments') }}
 
),
 
fine come (
 
   select
       ordini.order_id,
       ordini.customer_id,
       ordini.order_date,
       ordini.status,
       {% for payment_method in payment_methods -%}
       pagamenti_ordine.{{payment_method}}_amount,
       {% endfor -%}
       pagamenti_ordine.total_amount as amount
   from ordini
       left join pagamenti_ordine using (order_id)
 
)
 
select * from fine

Cosa interessante possiamo vedere qui?

In primo luogo: Sono stati utilizzati CTE (Common Table Expressions) — per organizzare e comprendere il codice, che contiene molte trasformazioni e logica aziendale

In secondo luogo: Il codice del modello è una miscela di SQL e Jinja linguaggio di templating.

Nell'esempio è stato utilizzato un ciclo per per calcolare l'importo per ciascun metodo di pagamento indicato nell'espressione set. Viene utilizzata anche la funzione ref — la possibilità di riferirsi all'interno del codice ad altri modelli:

Durante la compilazione ref sarà convertita in un puntatore di destinazione a una tabella o a una vista nel Data Warehouse
ref permette di costruire un grafo delle dipendenze dei modelli

Proprio Jinja aggiunge a DBT quasi illimitate capacità. Le più comuni sono:

If / else statements — operatori di diramazione
For loops — cicli
Variables — variabili
Macro — creazione di macro

Materializzazione: Tabella, Vista, Incrementale

La Strategia di Materializzazione — un approccio secondo cui il set di dati risultante del modello sarà salvato nello St storage.

In un'analisi di base, questo è:

Tabella — tabella fisica nello St storage
Vista — vista, tabella virtuale nello St storage

Ci sono anche strategie di materializzazione più complesse:

Incrementale — caricamento incrementale (di grandi tabelle di fatti); nuove righe vengono aggiunte, quelle modificate vengono aggiornate, quelle eliminate vengono rimosse
Ephemeral — il modello non viene materializzato direttamente, ma partecipa come CTE in altri modelli
Qualsiasi altra strategia che puoi aggiungere autonomamente

In aggiunta alle strategie di materializzazione, si aprono opportunità per ottimizzazioni specifiche per determinati St storage, ad esempio:

Snowflake: Tabelle transienti, Comportamento di unione, Clustering delle tabelle, Copia dei permessi, Viste sicure
Redshift: Distkey, Sortkey (interleaved, compound), Viste di collegamento tardivo
BigQuery: Partizionamento e clustering delle tabelle, Comportamento di unione, Crittografia KMS, Etichette e Tag
Spark: Formato file (parquet, csv, json, orc, delta), partition_by, clustered_by, buckets, incremental_strategy

Attualmente sono supportati i seguenti archivi:

Postgres
Redshift
BigQuery
Snowflake
Presto (parzialmente)
Spark (parzialmente)
Microsoft SQL Server (adapter comunitari)

Miglioriamo il nostro modello:

Rendiamo il suo riempimento incrementale (Incremental)
Aggiungiamo chiavi di segmentazione e ordinamento per Redshift

-- Configurazione del modello: 
-- Riempimento incrementale, chiave unica per l'aggiornamento dei record (unique_key)
-- Chiave di segmentazione (dist), chiave di ordinamento (sort)
{{
  config(
       materialized='incremental',
       unique_key='order_id',
       dist="customer_id",
       sort="order_date"
   )
}}
 
{% set payment_methods = ['credit_card', 'coupon', 'bank_transfer', 'gift_card'] %}
 
with orders as (
 
   select * from {{ ref('stg_orders') }}
   where 1=1
   {% if is_incremental() -%}
       -- Questo filtro sarà applicato solo per l'esecuzione incrementale
       and order_date >= (select max(order_date) from {{ this }})
   {%- endif %} 
 
),
 
order_payments as (
 
   select * from {{ ref('order_payments') }}
 
),
 
final as (
 
   select
       orders.order_id,
       orders.customer_id,
       orders.order_date,
       orders.status,
       {% for payment_method in payment_methods -%}
       order_payments.{{payment_method}}_amount,
       {% endfor -%}
       order_payments.total_amount as amount
   from orders
       left join order_payments using (order_id)
 
)
 
select * from final

Grafico delle dipendenze dei modelli

È anche l'albero delle dipendenze. È un DAG (Directed Acyclic Graph — Grafo Diretto A Cicli).

DBT costruisce il grafo sulla base della configurazione di tutti i modelli del progetto, in particolare dei collegamenti ref() all'interno dei modelli verso altri modelli. La presenza del grafo consente di fare quanto segue:

Eseguire i modelli nell'ordine corretto
Parallelizzazione della creazione delle vetrine
Eseguire un sottografo a scelta

Esempio di visualizzazione del grafo:

Ogni nodo del grafo rappresenta un modello, mentre i bordi del grafo sono definiti dall'espressione ref.

Qualità dei dati e Documentazione

Oltre a creare i modelli stessi, DBT consente di testare una serie di ipotesi (assertions) sul set di dati risultante, come ad esempio:

Not Null
Unique
Reference Integrity — integrità referenziale (ad esempio, customer_id nella tabella ordini corrisponde a id nella tabella clienti)
Corrispondenza con un elenco di valori consentiti

È possibile aggiungere i propri test (custom data tests), come, ad esempio, % di deviazione del fatturato rispetto ai dati di un giorno, una settimana, un mese fa. Qualsiasi ipotesi formulata sotto forma di query SQL può diventare un test.

In questo modo è possibile catturare nelle vetrine del Repository deviazioni indesiderate e errori nei dati.

Per quanto riguarda la documentazione, DBT fornisce meccanismi per aggiungere, versionare e distribuire metadati e commenti a livello di modelli e persino di attributi.

Ecco come appare l'aggiunta di test e documentazione a livello del file di configurazione:

 - name: fct_orders
   description: Questa tabella contiene informazioni di base sugli ordini, oltre ad alcuni fatti derivati basati sui pagamenti
   columns:
     - name: order_id
       tests:
         - unique # verifica l'unicità dei valori
         - not_null # verifica la presenza di null
       description: Questo è un identificatore unico per un ordine
     - name: customer_id
       description: Chiave estera alla tabella clienti
       tests:
         - not_null
         - relationships: # verifica l'integrità referenziale
             to: ref('dim_customers')
             field: customer_id
     - name: order_date
       description: Data (UTC) in cui è stato effettuato l'ordine
     - name: status
       description: '{{ doc("orders_status") }}'
       tests:
         - accepted_values: # verifica i valori accettabili
             values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']

Ecco come appare questa documentazione già sul sito web generato:

Macro e Moduli

L'obiettivo di DBT non è solo quello di diventare un insieme di script SQL, ma di fornire agli utenti strumenti potenti e ricchi di funzionalità per creare le proprie trasformazioni e distribuire questi moduli.

I macro sono insiemi di costrutti ed espressioni che possono essere richiamati come funzioni all'interno dei modelli. I macro consentono di riutilizzare SQL tra modelli e progetti in conformità con il principio dell'ingegneria DRY (Don't Repeat Yourself).

Esempio di macro:

{% macro rename_category(column_name) %}
case
 when {{ column_name }} ilike '%osx%' then 'osx'
 when {{ column_name }} ilike '%android%' then 'android'
 when {{ column_name }} ilike '%ios%' then 'ios'
 else 'other'
end as renamed_product
{% endmacro %}

E il suo utilizzo:

{% set column_name = 'product' %}
select
 product,
 {{ rename_category(column_name) }} -- richiamo del macro
from my_table

DBT viene fornito con un gestore pacchetti (packages) che consente agli utenti di pubblicare e riutilizzare singoli moduli e macro.

Questo significa la possibilità di caricare e utilizzare librerie come:

dbt_utils: gestione di Date/Time, Surrogate Keys, Schema tests, Pivot/Unpivot e altro
Template pronti per servizi come Snowplow e Stripe
Librerie per specifici Data Warehouse, ad esempio Redshift
Logging — Modulo per il logging delle attività di DBT

Puoi trovare l'elenco completo dei pacchetti su dbt hub.

Ancora più possibilità

Qui descriverò alcune altre caratteristiche interessanti e realizzazioni che io e il mio team utilizziamo per costruire un Data Warehouse in Wheely.

Separazione degli ambienti di esecuzione DEV — TEST — PROD

Anche all'interno di un unico cluster DWH (in diverse schemi). Ad esempio, utilizzando la seguente espressione:

with source as (
 
   select * from {{ source('salesforce', 'users') }}
   where 1=1
   {%- if target.name in ['dev', 'test', 'ci'] -%}           
       where timestamp >= dateadd(day, -3, current_date)   
   {%- endif -%}
 
)

Questo codice dice letteralmente: per gli ambienti dev, test, ci prendi i dati solo degli ultimi 3 giorni e non di più. Ciò significa che l'esecuzione in questi ambienti sarà molto più veloce e richiederà meno risorse. Quando viene eseguito nell'ambiente prod la condizione di filtro sarà ignorata.

Materializzazione con codifica alternativa delle colonne

Redshift è un DBMS a colonne che consente di impostare algoritmi di compressione dei dati per ogni singola colonna. La scelta degli algoritmi ottimali può ridurre il volume occupato su disco del 20-50%.

Macro redshift.compress_table eseguirà il comando ANALYZE COMPRESSION, creerà una nuova tabella con gli algoritmi di codifica raccomandati per le colonne, utilizzando le chiavi di segmentazione (dist_key) e ordinamento (sort_key), trasferirà i dati in essa e, se necessario, eliminerà la vecchia copia.

Firma del macro:

{{ compress_table(schema, table,
                 drop_backup=False,
                 comprows=none|Integer,
                 sort_style=none|compound|interleaved,
                 sort_keys=none|List,
                 dist_style=none|all|even,
                 dist_key=none|String) }}

Registrazione delle esecuzioni dei modelli

Per ogni esecuzione del modello possono essere associati hook che verranno eseguiti prima dell'avvio o subito dopo il completamento della creazione del modello:

   pre-hook: "{{ logging.log_model_start_event() }}"
   post-hook: "{{ logging.log_model_end_event() }}"

Il modulo di registrazione consentirà di registrare tutti i metadati necessari in una tabella separata, che in seguito può essere utilizzata per l'auditing e l'analisi delle problematiche (bottlenecks).

Ecco come appare il dashboard sui dati di registrazione in Looker:

Automazione della manutenzione dello Storage

Se utilizzi estensioni delle funzionalità dello Storage utilizzato, come le UDF (Funzioni definite dall'utente), la gestione delle versioni di queste funzioni, il controllo degli accessi e il rilascio automatizzato di nuove versioni è molto comodo da realizzare in DBT.

Noi utilizziamo UDF in Python per calcolare hash, domini degli indirizzi email e decodificare maschere di bit (bitmask).

Esempio di un macro che crea una UDF in qualsiasi ambiente di esecuzione (sviluppo, test, produzione):

{% macro create_udf() -%}
  
  {% set sql %}
        CREATE OR REPLACE FUNCTION {{ target.schema }}.f_sha256(mes "varchar")
            RETURNS varchar
            LANGUAGE plpythonu
            STABLE
        AS $$  
            import hashlib
            return hashlib.sha256(mes).hexdigest()
        $$
        ;
  {% endset %}
  
  {% set table = run_query(sql) %}
  
{%- endmacro %}

In Wheely utilizziamo Amazon Redshift, che si basa su PostgreSQL. Per Redshift è importante raccogliere regolarmente statistiche sulle tabelle e liberare spazio su disco — i comandi ANALYZE e VACUUM, rispettivamente.

A tal fine, ogni notte vengono eseguiti i comandi del macro redshift_maintenance:

{% macro redshift_maintenance() %}
 
   {% set vacuumable_tables=run_query(vacuumable_tables_sql) %}
 
   {% for row in vacuumable_tables %}
       {% set message_prefix=loop.index ~ " di " ~ loop.length %}
 
       {%- set relation_to_vacuum = adapter.get_relation(
                                           database=row['table_database'],
                                           schema=row['table_schema'],
                                           identifier=row['table_name']
                                   ) -%}
       {% do run_query("commit") %}
 
       {% if relation_to_vacuum %}
           {% set start=modules.datetime.datetime.now() %}
           {{ dbt_utils.log_info(message_prefix ~ " Vacuuming " ~ relation_to_vacuum) }}
           {% do run_query("VACUUM " ~ relation_to_vacuum ~ " BOOST") %}
           {{ dbt_utils.log_info(message_prefix ~ " Analizzando " ~ relation_to_vacuum) }}
           {% do run_query("ANALYZE " ~ relation_to_vacuum) %}
           {% set end=modules.datetime.datetime.now() %}
           {% set total_seconds = (end - start).total_seconds() | round(2)  %}
           {{ dbt_utils.log_info(message_prefix ~ " Completato " ~ relation_to_vacuum ~ " in " ~ total_seconds ~ "s") }}
       {% else %}
           {{ dbt_utils.log_info(message_prefix ~ ' Saltando la relazione "' ~ row.values() | join ('"."') ~ '" poiché non esiste') }}
       {% endif %}
 
   {% endfor %}
 
{% endmacro %}

DBT Cloud

È possibile utilizzare DBT come servizio (Managed Service). Incluso:

Web IDE per lo sviluppo di progetti e modelli
Configurazione dei job e programmazione
Accesso semplice e comodo ai log
Sito web con la documentazione del tuo progetto
Integrazione CI (Continuous Integration)

Conclusione

Preparare e utilizzare DWH diventa piacevole e benefico come bere un frullato. DBT è composto da Jinja, estensioni personalizzate (moduli), compilatore, motore (executor) e gestore dei pacchetti. Riunendo questi elementi, ottieni un ambiente di lavoro completo per il tuo Data Warehouse. Oggi non esiste un modo migliore per gestire le trasformazioni all'interno del DWH.

Le convinzioni seguite dai sviluppatori di DBT sono formulate così:

Il codice, e non il GUI, è la migliore astrazione per esprimere logica analitica complessa
Lavorare con i dati dovrebbe adattare le migliori pratiche nello sviluppo software (Software Engineering)

L'infrastruttura chiave per la gestione dei dati dovrebbe essere controllata dalla comunità degli utenti come software open source
Non solo gli strumenti di analisi, ma anche il codice sarà sempre più patrimonio della comunità Open Source

Queste convinzioni fondamentali hanno dato vita a un prodotto utilizzato oggi da oltre 850 aziende, e costituiscono la base per molte interessanti estensioni che saranno sviluppate in futuro.

Per chi fosse interessato, c'è la registrazione video di una lezione aperta che ho tenuto alcune settimane fa nel contesto di una lezione aperta in OTUS — Data Build Tool per il data warehouse Amazon Redshift.

Oltre a DBT e ai Data Warehouse, nel corso di Data Engineer sulla piattaforma OTUS, io e i miei colleghi teniamo corsi su una serie di altri argomenti rilevanti e moderni: