Su quali principi si basa un ideale Data Warehouse?

Focus sul valore aziendale e sull'analisi in assenza di boilerplate code. Gestire il DWH come una base di codice: versionamento, revisione, test automatici e CI. Modularità, estensibilità, open source e comunità. Documentazione utente amichevole e visualizzazione delle dipendenze (Data Lineage).

Tutti questi argomenti in dettaglio e il ruolo di DBT nell'ecosistema Big Data & Analytics — benvenuti sotto il post.

Ciao a tutti

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

Panoramica

Il framework DBT riguarda tutta la lettera T nell'acronimo ELT (Extract — Transform — Load).

Con l'emergere di database analitici così performanti e scalabili come BigQuery, Redshift, Snowflake, è venuto a mancare qualsiasi senso nel fare trasformazioni al di fuori del Data Warehouse.

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

La funzione principale di DBT è prendere il codice, compilarlo in SQL e eseguire i comandi nella giusta sequenza nel Warehouse.

Struttura del progetto DBT

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

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

A livello base, il lavoro viene svolto come segue:

L'utente prepara il codice dei modelli in qualsiasi IDE comoda
Utilizzando la CLI, viene avviata l'esecuzione dei modelli, DBT compila il codice dei modelli in SQL
Il codice SQL compilato viene eseguito nel Warehouse nell'ordine specificato (grafo)

Ecco come potrebbe apparire l'esecuzione 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 vostre query nel Warehouse (variazioni dei comandi CREATE, INSERT, UPDATE, DELETE ALTER, GRANT, ...).

Qualsiasi modello implica la scrittura di una query SELECT che determina il set di dati risultante.

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

{% set payment_methods = ['credit_card', 'coupon', 'bank_transfer', 'gift_card'] %}
 
with orders as (
 
   select * from {{ ref('stg_orders') }}
 
),
 
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

Cosa interessante possiamo vedere qui?

In primo luogo: Utilizzo delle CTE (Common Table Expressions) — per organizzare e comprendere un codice che contiene molte trasformazioni e logica del business

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

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

Durante la compilazione ref verrà convertito in un puntatore di destinazione a una tabella o vista nel magazzino.
ref permette di costruire un grafo delle dipendenze dei modelli.

È proprio questo Jinja che dà a DBT praticamente illimitate possibilità. Le più comunemente utilizzate includono:

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

Materializzazione: Table, View, Incremental

La Strategia di Materializzazione — un approccio secondo cui il set di dati risultante del modello verrà salvato nel magazzino.

In una visione di base questo è:

Table — tabella fisica nel magazzino
View — vista, tabella virtuale nel magazzino

Ci sono anche strategie di materializzazione più complesse:

Incremental — 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 tu stesso.

In aggiunta alle strategie di materializzazione, si aprono opportunità di ottimizzazione per specifici magazzini, ad esempio:

Snowflake: Transient tables, Merge behavior, Table clustering, Copying grants, Secure views
Redshift: Distkey, Sortkey (interleaved, compound), Late Binding Views
BigQuery: Partitioning & clustering delle tabelle, Merge behavior, KMS Encryption, Labels & Tags
Spark: Formato file (parquet, csv, json, orc, delta), partition_by, clustered_by, buckets, strategia_incremenale

Attualmente sono supportati i seguenti Storage:

Postgres
Redshift
BigQuery
Snowflake
Presto (parzialmente)
Spark (parzialmente)
Microsoft SQL Server (adattatore community)

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

Grafo delle dipendenze dei modelli

Chiamato anche albero delle dipendenze. È un DAG (Directed Acyclic Graph — Grafo Afinito Diretto).

DBT costruisce un grafo basato sulla configurazione di tutti i modelli del progetto, più precisamente sui riferimenti ref() all'interno dei modelli vers altri modelli. La presenza del grafo consente le seguenti operazioni:

Esecuzione dei modelli in sequenza corretta
Parallelizzazione della creazione dei data mart
Esecuzione di un sottografo arbitrario

Esempio di visualizzazione del grafo:

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

Qualità dei dati e Documentazione

Oltre alla creazione dei modelli stessi, DBT consente di testare una serie di assunzioni (assertions) sul set di dati risultante, come ad esempio:

Not Null
Unique
Reference Integrity — integrità referenziale (ad esempio, customer_id nella tabella orders corrisponde a id nella tabella customers)
Corrispondenza con l'elenco dei valori consentiti

È possibile aggiungere test personalizzati (custom data tests), ad esempio la % di deviazione delle entrate rispetto ai dati di un giorno, una settimana, un mese fa. Qualsiasi assunzione formulata sotto forma di query SQL può diventare un test.

In questo modo è possibile rilevare deviazioni indesiderate e errori nei dati nei data mart dello Storage.

Per quanto riguarda la documentazione, DBT fornisce meccanismi per l'aggiunta, la versioning e la diffusione di 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 # controllo sull'unicità dei valori
         - not_null # controllo sulla presenza di null
       description: Questo è un identificatore unico per un ordine
     - name: customer_id
       description: Chiave esterna alla tabella dei clienti
       tests:
         - not_null
         - relationships: # controllo sull'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: # controllo sui valori accettabili
             values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']

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

Macro e Moduli

Lo scopo di DBT non è tanto quello di diventare un insieme di script SQL, ma fornire agli utenti strumenti potenti e ricchi di funzionalità per costruire le proprie trasformazioni e distribuire questi moduli.

Le macro sono insiemi di costrutti ed espressioni che possono essere richiamati come funzioni all'interno dei modelli. Le macro consentono di riutilizzare SQL tra modelli e progetti secondo il principio ingegneristico 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) }} -- chiamata alla macro
from my_table

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

Ciò significa la possibilità di caricare e utilizzare librerie come:

dbt_utils: operazioni su Date/Time, Chiavi Surrogate, Test di Schema, Pivot/Unpivot e altro
Modelli predefiniti per servizi come Snowplow e Stripe
Librerie per specifici Data Warehouse, ad esempio Redshift
Registrazione — Modulo per il logging del funzionamento di DBT

L'elenco completo dei pacchetti è disponibile su dbt hub.

Ancora più funzionalità

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

Separazione degli ambienti DEV — TEST — PROD

Anche all'interno di un singolo cluster DWH (all'interno di diversi 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ù. Quindi, l'esecuzione in questi ambienti sarà molto più veloce e richiederà meno risorse. Quando viene eseguita nell'ambiente prod la condizione di filtro sarà ignorata.

Materializzazione con codifica alternativa delle colonne

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

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

Firma della 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

A ogni esecuzione di un modello è possibile collegare hook che verranno eseguiti prima dell'avvio o subito dopo la fine 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, sulla quale successivamente sarà possibile effettuare audit e analisi dei punti critici (bottlenecks).

Ecco come appare il dashboard con i dati di registrazione in Looker:

Automazione della manutenzione del Data Warehouse

Se utilizzi delle funzionalità estese del Data Warehouse utilizzato, come le UDF (User Defined Functions), il versionamento di queste funzioni, la gestione degli accessi e la distribuzione automatizzata di nuove release è molto comodo effettuarlo in DBT.

Utilizziamo UDF in Python per calcolare valori hash, domini di indirizzi email e decodificare le maschere bit (bitmask).

Ecco un esempio di macro che crea UDF in qualsiasi ambiente di esecuzione (dev, test, prod):

{% 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. È importante raccogliere regolarmente le statistiche delle tabelle e liberare spazio su disco — i comandi ANALYZE e VACUUM, rispettivamente.

A tal fine, ogni notte vengono eseguiti i comandi della 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 ~ " Pulizia di " ~ 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 ~ ' Salto della relazione "' ~ row.values() | join ('"."') ~ '" poiché non esiste') }}
 {% endif %}
 
 {% endfor %}
 
{% endmacro %}

DBT Cloud

È possibile utilizzare DBT come servizio (Servizio Gestito). Compreso:

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

Conclusione

Cucinare e utilizzare DWH diventa altrettanto piacevole e benefico quanto bere frullati. DBT è composto da Jinja, estensioni personalizzate (moduli), un compilatore, un motore (executor) e un gestore di pacchetti. Combinando questi elementi, ottieni un ambiente di lavoro completo per il tuo Data Warehouse. Oggi è difficile trovare un modo migliore per gestire le trasformazioni all'interno di DWH.

Le convinzioni seguite dagli sviluppatori di DBT sono formulate come segue:

Il codice, e non l'interfaccia grafica, è la migliore astrazione per esprimere logiche analitiche complesse.
Il lavoro con i dati deve adattare le migliori pratiche di sviluppo software (Software Engineering).

L'infrastruttura fondamentale per il lavoro con i dati deve essere gestita dalla comunità degli utenti come software open source.
Non solo gli strumenti di analisi, ma anche il codice diventerà 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 verranno create in futuro.

Per coloro che sono interessati, è disponibile la registrazione video di una lezione aperta che ho tenuto alcuni mesi fa nell'ambito della 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 trattiamo una serie di altri temi attuali e moderni:

Concetti architettonici delle applicazioni di Big Data.
Pratica con Spark e Spark Streaming.
Studio dei metodi e degli strumenti per il caricamento delle fonti di dati.
Costruzione di vetrine analitiche in DWH.
Concetti NoSQL: HBase, Cassandra, ElasticSearch.
Principi di organizzazione del monitoraggio e dell'orchestrazione.
Progetto finale: mettiamo insieme tutte le competenze sotto la guida di un mentore.