Auf welchen Prinzipien basiert ein ideales Data Warehouse?

Fokus auf Geschäftswert und Analytik bei Abwesenheit von Boilerplate-Code. Management von DWH wie eine Codebasis: Versionskontrolle, Reviews, automatisierte Tests und CI. Modularität, Erweiterbarkeit, Open Source und Community. Benutzerfreundliche Dokumentation und Visualisierung von Abhängigkeiten (Data Lineage).

Darüber hinaus und über die Rolle von DBT im Ökosystem Big Data & Analytics — herzlich willkommen unter dem Artikel.

Hallo zusammen

Hier ist Artemiy Kozyr. Ich arbeite seit über 5 Jahren mit Data Warehouses, beschäftige mich mit ETL/ELT und ebenso mit Datenanalytik und Visualisierung. Momentan arbeite ich bei Wheely, lehre bei OTUS im Kurs Data Engineer, und heute möchte ich einen Artikel mit euch teilen, den ich zur Eröffnung eines neuen Kurses.

Kurze Übersicht

Das DBT-Framework dreht sich um den Buchstaben T im Akronym ELT (Extract — Transform — Load).

Mit dem Aufkommen leistungsstarker und skalierbarer Analytics-Datenbanken wie BigQuery, Redshift, Snowflake, hat es keinen Sinn mehr, Transformierungen außerhalb des Data Warehouses durchzuführen.

DBT lädt keine Daten aus Quellen, sondern bietet enorme Möglichkeiten zur Arbeit mit den Daten, die bereits im Warehouse (in Internal oder External Storage) geladen sind.

Die Hauptaufgabe von DBT ist es, den Code zu nehmen, ihn in SQL zu kompilieren und die Befehle in der richtigen Reihenfolge im Warehouse auszuführen.

Struktur eines DBT-Projekts

Ein Projekt besteht aus zwei Arten von Verzeichnissen und Dateien:

Modell (.sql) — eine Transformationseinheit, ausgedrückt durch eine SELECT-Anfrage
Konfigurationsdatei (.yml) — Parameter, Einstellungen, Tests, Dokumentation

Auf einer grundlegenden Ebene läuft die Arbeit folgendermaßen ab:

Der Benutzer bereitet den Code der Modelle in einer beliebigen geeigneten IDE vor
Über die CLI wird der Start der Modelle aufgerufen, DBT kompiliert den Code der Modelle in SQL
Der kompilierte SQL-Code wird im Warehouse in der vorgegebenen Reihenfolge (Graph) ausgeführt

So könnte der Start über die CLI aussehen:

Alles ist SELECT

Das ist das Killer-Feature des Data Build Tool-Frameworks. Mit anderen Worten, DBT abstrahiert den gesamten Code, der mit der Materialisierung Ihrer Abfragen im Warehouse verbunden ist (Variationen von CREATE, INSERT, UPDATE, DELETE, ALTER, GRANT, …).

Jedes Modell setzt das Schreiben einer SELECT-Anfrage voraus, die den resultierenden Datensatz definiert.

Die Logik der Transformationen kann mehrstufig sein und Daten aus mehreren anderen Modellen konsolidieren. Ein Beispiel für ein Modell, das eine Auftragsübersicht (f_orders) erstellt, ist:

{% 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

Was Interessantes können wir hier sehen?

Zunächst einmal: Es wurden CTE (Common Table Expressions) verwendet — um den Code zu organisieren und zu verstehen, der viele Transformationen und Geschäftslogik enthält.

Zweitens: Der Modellcode ist eine Mischung aus SQL und Jinja (Template-Sprache).

Im Beispiel wird eine Schleife verwendet for um die Beträge für jede der im Ausdruck angegebenen Zahlungsmethoden zu erstellen. set. Auch die Funktion ref wird verwendet — die Möglichkeit, innerhalb des Codes auf andere Modelle zu verweisen:

Während der Kompilierung ref wird in einen Zielverweis auf eine Tabelle oder Sicht im Speicher umgewandelt.
ref Sie ermöglicht den Aufbau eines Abhängigkeitsdiagramms von Modellen.

Genau das Jinja verleiht DBT nahezu unbegrenzte Möglichkeiten. Die häufigsten davon sind:

If / else-Anweisungen — Verzweigungsoperatoren.
For-Schleifen — Schleifen.
Variablen — Variablen.
Makro — Erstellung von Makros.

Materialisierung: Tabelle, Sicht, Inkremental.

Die Materialisierungsstrategie — ein Ansatz, bei dem der resultierende Datensatz eines Modells im Speicher gespeichert wird.

Im grundlegenden Sinne lautet dies:

Tabelle — physische Tabelle im Speicher.
Sicht — Sicht, virtuelle Tabelle im Speicher.

Es gibt auch komplexere Materialisierungsstrategien:

Inkremental — inkrementelles Laden (großer Faktentabellen); neue Zeilen werden hinzugefügt, geänderte Zeilen aktualisiert, gelöschte Zeilen entfernt.
Ephemeral — das Modell wird nicht direkt materialisiert, sondern als CTE in anderen Modellen verwendet.
Sonstige Strategien, die Sie selbst hinzufügen können.

Zusätzlich zu den Materialisierungsstrategien eröffnen sich Möglichkeiten zur Optimierung für bestimmte Speicherlösungen, wie z.B.:

Snowflake: Transiente Tabellen, Merge-Verhalten, Tabellenklusterung, Grant-Kopieren, sichere Sichten.
Redshift: Distkey, Sortkey (interleaved, compound), Late Binding Views.
BigQuery: Tabellenpartitionierung & Klusterung, Merge-Verhalten, KMS-Verschlüsselung, Labels & Tags.
Spark: Dateiformat (parquet, csv, json, orc, delta), partition_by, clustered_by, buckets, incremental_strategy

Aktuell werden folgende Speicher unterstützt:

Postgres
Redshift
BigQuery
Snowflake
Presto (teilweise)
Spark (teilweise)
Microsoft SQL Server (Community-Adapter)

Lassen Sie uns unser Modell verbessern:

Machen wir die Befüllung inkrementell (Incremental)
Fügen wir Segmentierungs- und Sortierschlüssel für Redshift hinzu

-- Modellkonfiguration: 
-- Inkrementelle Befüllung, eindeutiger Schlüssel zum Aktualisieren von Datensätzen (unique_key)
-- Segmentierungsschlüssel (dist), Sortierschlüssel (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() -%}
       -- Dieser Filter wird nur für inkrementelle Ausführung angewendet
       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

Modellabhängigkeiten-Diagramm

Es ist auch als Abhängigkeitsbaum bekannt. Auch DAG (Directed Acyclic Graph — Richtungsloser Acyclic Graph).

DBT erstellt ein Diagramm basierend auf der Konfiguration aller Modelle des Projekts, genauer gesagt auf den ref()-Verweisen innerhalb der Modelle auf andere Modelle. Das Vorhandensein eines Diagramms ermöglicht folgende Dinge:

Ausführung der Modelle in der richtigen Reihenfolge
Parallelisierung der Erstellung von Datenansichten
Ausführung eines beliebigen Teil-Diagramms

Beispiel für die Visualisierung des Diagramms:

Jeder Knoten des Diagramms ist ein Modell, die Kanten des Diagramms werden durch den Ausdruck ref definiert.

Datenqualität und Dokumentation

Neben der Erstellung der Modelle ermöglicht DBT, eine Reihe von Annahmen (Assertions) über den resultierenden Datensatz zu testen, wie z.B.:

Not Null
Eindeutig
Referenzintegrität — referentielle Integrität (z.B. customer_id in der Tabelle orders entspricht id in der Tabelle customers)
Übereinstimmung mit einer Liste zulässiger Werte

Es besteht die Möglichkeit, eigene Tests (custom data tests) hinzuzufügen, wie z.B. den % Abweichung des Umsatzes im Vergleich zu den Werten vor einem Tag, einer Woche oder einem Monat. Jede Annahme, die in Form eines SQL-Abfrage formuliert ist, kann zu einem Test werden.

Auf diese Weise können unerwünschte Abweichungen und Fehler in den Daten in den Datenansichten des Speichers erfasst werden.

Was die Dokumentation betrifft, bietet DBT Mechanismen zum Hinzufügen, Versionieren und Verteilen von Metadaten und Kommentaren auf Modellebene und sogar auf Attributebene.

So sieht das Hinzufügen von Tests und Dokumentation auf der Ebene der Konfigurationsdatei aus:

 - name: fct_orders
   description: Diese Tabelle enthält grundlegende Informationen zu Bestellungen sowie einige abgeleitete Fakten basierend auf Zahlungen
   columns:
     - name: order_id
       tests:
         - unique # Überprüfung auf Eindeutigkeit der Werte
         - not_null # Überprüfung auf Vorhandensein von null
       description: Dies ist ein eindeutiger Identifikator für eine Bestellung
     - name: customer_id
       description: Fremdschlüssel zur Kundentabelle
       tests:
         - not_null
         - relationships: # Überprüfung der referenziellen Integrität
             to: ref('dim_customers')
             field: customer_id
     - name: order_date
       description: Datum (UTC), an dem die Bestellung aufgegeben wurde
     - name: status
       description: '{{ doc("orders_status") }}'
       tests:
         - accepted_values: # Überprüfung auf zulässige Werte
             values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']

So sieht diese Dokumentation bereits auf der generierten Webseite aus:

Makros und Module

Ziel von DBT ist es nicht so sehr, eine Sammlung von SQL-Skripten zu sein, sondern den Benutzern leistungsstarke und funktional reiche Werkzeuge zur Verfügung zu stellen, um eigene Transformationen zu erstellen und diese Module zu verteilen.

Makros sind Sets von Konstruktionen und Ausdrücken, die als Funktionen innerhalb von Modellen aufgerufen werden können. Makros ermöglichen die Wiederverwendung von SQL zwischen Modellen und Projekten gemäß dem Ingenieurprinzip DRY (Don’t Repeat Yourself).

Beispiel eines Makros:

{% 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 %}

Und seine Verwendung:

{% set column_name = 'product' %}
select
 product,
 {{ rename_category(column_name) }} -- Aufruf des Makros
from my_table

DBT wird mit einem Paketmanager (packages) geliefert, der es Benutzern ermöglicht, einzelne Module und Makros zu veröffentlichen und wiederzuverwenden.

Dies bedeutet die Möglichkeit, Bibliotheken wie diese herunterzuladen und zu verwenden:

dbt_utils: Arbeiten mit Datum/Uhrzeit, Surrogatschlüsseln, Schema-Tests, Pivot/Unpivot und anderen
Fertige Vorlagen für Data Warehouses von Anbietern wie Snowplow und Stripe
Bibliotheken für bestimmte Data Warehouses, z.B. Redshift
Logging — Modul zur Protokollierung der DBT-Arbeiten

Eine vollständige Liste der Pakete finden Sie unter dbt hub.

Noch mehr Möglichkeiten

Hier beschreibe ich einige weitere interessante Besonderheiten und Implementierungen, die ich und mein Team für den Aufbau des Data Warehouses verwenden. Wheely.

Trennung der Ausführungsumgebungen DEV — TEST — PROD

Selbst innerhalb eines DWH-Clusters (innerhalb verschiedener Schemata). Zum Beispiel mit folgender Anweisung:

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 -%}
 
)

Dieser Code sagt im Grunde: für die Umgebungen dev, test, ci nimm nur die Daten der letzten 3 Tage, nicht mehr. Das heißt, die Ausführung in diesen Umgebungen wird viel schneller und erfordert weniger Ressourcen. Bei der Ausführung in der Umgebung prod wird die Filterbedingung ignoriert.

Materialisierung mit alternativer Spaltenkodierung

Redshift ist eine spaltenorientierte Datenbank, die es ermöglicht, Kompressionsalgorithmen für jede einzelne Spalte festzulegen. Die Auswahl optimaler Algorithmen kann den benötigten Speicherplatz um 20-50 % reduzieren.

Makro redshift.compress_table führt den Befehl ANALYZE COMPRESSION aus, erstellt eine neue Tabelle mit den empfohlenen Kodierungsalgorithmen für die Spalten, den angegebenen Segmentierungsschlüsseln (dist_key) und Sortierschlüsseln (sort_key), überträgt die Daten dorthin und löscht bei Bedarf die alte Kopie.

Signatur des Makros:

{{ 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) }}

Protokollierung von Modellausführungen

An jede Ausführung des Modells können Hooks (Hooks) angehängt werden, die vor der Ausführung oder sofort nach dem Abschluss der Modellerstellung ausgeführt werden:

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

Das Protokollierungsmodul ermöglicht es, alle erforderlichen Metadaten in einer separaten Tabelle festzuhalten, anhand derer später eine Prüfung und Analyse von Engpässen (bottlenecks) durchgeführt werden kann.

So sieht das Dashboard mit den Protokolldaten in Looker aus:

Automatisierung der Wartung des Data Warehouses

Wenn Sie erweiterte Funktionen des verwendeten Data Warehouses wie UDF (User Defined Functions) nutzen, ist das Versionieren dieser Funktionen, das Verwalten von Berechtigungen und das automatisierte Veröffentlichen neuer Versionen in DBT sehr praktisch.

Wir verwenden UDF in Python zur Berechnung von Hash-Werten, E-Mail-Domänen und zur Dekodierung von Bitmasken.

Beispiel eines Makros, das UDF in jeder Ausführungsumgebung erstellt (Entwicklung, Test, Produktion):

{% 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 %}

Bei Wheely verwenden wir Amazon Redshift, das auf PostgreSQL basiert. Für Redshift ist es wichtig, regelmäßig Statistiken zu den Tabellen zu sammeln und Speicherplatz auf der Festplatte freizugeben – die Befehle ANALYZE und VACUUM, entsprechend.

Dafür werden jede Nacht die Befehle aus dem Makro redshift_maintenance ausgeführt:

{% macro redshift_maintenance() %}
 
 {% set vacuumable_tables=run_query(vacuumable_tables_sql) %}
 
 {% for row in vacuumable_tables %}
 {% set message_prefix=loop.index ~ " von " ~ 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 ~ " Vacuumieren " ~ relation_to_vacuum) }}
 {% do run_query("VACUUM " ~ relation_to_vacuum ~ " BOOST") %}
 {{ dbt_utils.log_info(message_prefix ~ " Analysieren " ~ 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 ~ " Fertig " ~ relation_to_vacuum ~ " in " ~ total_seconds ~ "s") }}
 {% else %}
 {{ dbt_utils.log_info(message_prefix ~ ' Überspringe Relation "' ~ row.values() | join ('"."') ~ '" da sie nicht existiert') }}
 {% endif %}
 
 {% endfor %}
 
{% endmacro %}

DBT Cloud

Es besteht die Möglichkeit, DBT als Dienst (Managed Service) zu nutzen. Im Paket enthalten:

Web-IDE zur Entwicklung von Projekten und Modellen
Konfiguration von Jobs und Zeitplanung
Einfacher und bequemer Zugang zu Logs
Webseite mit der Dokumentation Ihres Projekts
Anbindung von CI (Continuous Integration)

Fazit

Es macht genauso viel Freude und ist ebenso wohltuend, DWH zu erstellen und zu verwenden, wie Smoothies zu trinken. DBT besteht aus Jinja, benutzerdefinierten Erweiterungen (Modulen), einem Compiler, einem Executor und einem Paketmanager. Indem Sie diese Elemente zusammenfügen, schaffen Sie eine vollständige Arbeitsumgebung für Ihr Data Warehouse. Heute gibt es kaum einen besseren Weg, Transformationen innerhalb von DWH zu verwalten.

Die Überzeugungen, denen die DBT-Entwickler folgten, lassen sich so formulieren:

Code, nicht GUI, ist die beste Abstraktion zur Darstellung komplexer analytischer Logik.
Die Arbeit mit Daten sollte bewährte Praktiken der Softwareentwicklung (Software Engineering) anpassen.

Die wichtigste Infrastruktur zur Datenverarbeitung sollte von der Benutzer-Community wie Open-Source-Software kontrolliert werden.
Nicht nur Analysewerkzeuge, sondern auch Code wird zunehmend zum Gemeingut der Open-Source-Community.

Diese grundlegenden Überzeugungen haben ein Produkt hervorgebracht, das heute in mehr als 850 Unternehmen verwendet wird, und sie bilden die Grundlage für viele interessante Erweiterungen, die in Zukunft geschaffen werden.

Für diejenigen, die interessiert sind, gibt es eine Videoaufzeichnung einer offenen Lektion, die ich vor einigen Monaten im Rahmen einer offenen Lektion bei OTUS gehalten habe — Data Build Tool für das Amazon Redshift Data Warehouse..

Neben DBT und Data Warehouses behandeln ich und meine Kollegen im Data Engineering-Kurs auf der OTUS-Plattform eine Reihe anderer aktueller und moderner Themen:

Architekturkonzepte für Big Data-Anwendungen.
Praxis mit Spark und Spark Streaming.
Untersuchung der Methoden und Werkzeuge zum Laden von Datenquellen.
Erstellung analytischer ODS (Operational Data Store) in DWH.
NoSQL-Konzepte: HBase, Cassandra, ElasticSearch.
Prinzipien der Organisation von Monitoring und Orchestrierung.
Abschlussprojekt: alle Fähigkeiten unter mentorieller Unterstützung zusammenführen.