На какви принципи се основава идеалното Хранилище на Данни?

Фокус върху бизнес стойността и аналитиката при отсъствие на boilerplate код. Управление на DWH като кодова база: версиониране, рецензиране, автоматизирано тестване и CI. Модуларност, разширяемост, отворен изходен код и общност. Удобна потребителска документация и визуализация на зависимостите (Data Lineage).

За всичко това по-подробно и за ролята на DBT в екосистемата на Big Data & Analytics – добре дошли под кат.

Здравейте на всички

Свързан съм Артемий Козырь. Вече повече от 5 години работя с хранилища на данни, занимавам се с изграждане на ETL/ELT, а също така и с анализ на данни и визуализация. В момента работя в Wheely, преподавам в OTUS на курса Data Engineer, и днес искам да споделя с вас статия, която написах в навечерието на старта на нов набор на курса.

Кратък обзор

Фреймворкът DBT – това е всичко за буквата T в акронима ELT (Extract – Transform – Load).

С появата на толкова производителни и мащабируеми аналитични бази данни като BigQuery, Redshift, Snowflake, изчезна смисълът да се правят трансформации извън Хранилището на Данни.

DBT не изтегля данни от източници, но предоставя огромни възможности за работа с данни, които вече са заредени в Хранилището (в Internal или External Storage).

Основната цел на DBT е да вземе кода, да го компилира в SQL и да изпълни командите в правилната последователност в Хранилището.

Структура на проекта DBT

Проектът се състои от две директории и файлове от само два типа:

Модел (.sql) – единица за трансформация, изразена с SELECT заявка
Файл с конфигурация (.yml) – параметри, настройки, тестове, документация

На основно ниво работата се изгражда по следния начин:

Потребителят подготвя кода на моделите в удобно IDE
С помощта на CLI се извиква стартиране на моделите, DBT компилира кода на моделите в SQL
Скомпилираният SQL код се изпълнява в Хранилището в зададена последователност (граф)

Ето как може да изглежда стартирането от CLI:

Всичко е SELECT

Това е килър-функцията на фреймворка Data Build Tool. С други думи, DBT абстрахира целия код, свързан с материализацията на вашите заявки в Хранилището (вариации от командите CREATE, INSERT, UPDATE, DELETE ALTER, GRANT, …).

Всяка модел предполага написването на една SELECT заявка, която определя резултатния набор от данни.

При това логиката на преобразуванията може да бъде многостепенна и да консолидира данни от няколко други модела. Пример за модел, който ще построи витрина за поръчки (f_orders):

{% set payment_methods = ['credit_card', 'coupon', 'bank_transfer', 'gift_card'] %}
 
с поръчки като (
 
   избери * от {{ ref('stg_orders') }}
 
),
 
плащания по поръчки като (
 
   избери * от {{ ref('order_payments') }}
 
),
 
финално като (
 
   избери
       поръчки.order_id,
       поръчки.customer_id,
       поръчки.order_date,
       поръчки.status,
       {% for payment_method in payment_methods -%}
       order_payments.{{payment_method}}_amount,
       {% endfor -%}
       order_payments.total_amount като amount
   от поръчки
       ляво съединение order_payments using (order_id)
 
)
 
избери * от финално

Какво интересно можем да видим тук?

На първо място: Използвани са CTE (Общи таблици за изразяване) — за организация и разбиране на кода, който съдържа много преобразувания и бизнес логика

На второ място: Кодът на модела — това е смесица от SQL и езика Jinja (език за шаблони).

В примера е използван цикъл for за формиране на сумата за всеки метод на плащане, указан в израза set. Също така се използва функцията ref — възможност да се позовеш вътре в кода на други модели:

По време на компилирането ref ще бъде преобразуван в целеви указател към таблица или представление в хранилището
ref позволява да се изгради граф на зависимостите на моделите

Точно това Jinja добавя в DBT почти неограничени възможности. Най-често използваните от тях:

If / else оператори — оператори за разклонение
For цикли — цикли
Променливи — променливи
Макрос — създаване на макроси

Материализация: Таблица, Представление, Инкрементално

Стратегия за материализация — подход, според който резултатният набор от данни на модела ще бъде запазен в хранилището.

В основния подход това е:

Таблица — физическа таблица в хранилището
Представление — виртуална таблица в хранилището

Има и по-сложни стратегии за материализация:

Инкрементално — инкрементално зареждане (на големи таблици с факти); нови редове се добавят, променените — обновяват, изтритите — изчистват
Ефимерно — моделът не се материализира директно, но участва като CTE в други модели
Всякакви други стратегии, които можете да добавите сами

В допълнение към стратегиите за материализация се отварят възможности за оптимизация за конкретни хранилища, например:

Snowflake: Преходни таблици, Merge поведение, Класифициране на таблици, Копиране на права, Сигурни представления
Redshift: Distkey, Sortkey (взаимно, комбинирано), Класически представления
BigQuery: Партициониране на таблици & класифициране, Merge поведение, KMS Шифроване, Етикети & Тагове
Spark: Формат на файла (parquet, csv, json, orc, delta), partition_by, clustered_by, buckets, incremental_strategy

Към момента се поддържат следните Хранилища:

Postgres
Redshift
BigQuery
Snowflake
Presto (частично)
Spark (частично)
Microsoft SQL Server (адаптер на общността)

Нека подобрим нашия модел:

Да направим попълването му инкрементално (Incremental)
Да добавим ключове за сегментиране и сортиране за Redshift

-- Конфигурация на модела:  
-- Инкрементално попълване, уникален ключ за обновление на записи (unique_key)
-- Ключ за сегментиране (dist), ключ за сортиране (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() -%}
       -- Този филтър ще бъде приложен само за инкрементални стартирания
       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

Граф на зависимостите на моделите

Той също е дърво на зависимостите. Той е DAG (Directed Acyclic Graph — Направен Ацикличен Граф).

DBT изгражда граф на базата на конфигурацията на всички модели в проекта, а именно референциите ref() в моделите към други модели. Наличието на граф позволява следните действия:

Стартиране на модели в правилната последователност
Паралелизация на формирането на витрини
Стартиране на произволен подграф

Пример за визуализация на графа:

Всеки връх на графа — това е модел, ребрата на графа се определят чрез израза ref.

Качество на данните и документация

Освен формирането на самите модели, DBT позволява тестване на редица предположения (assertions) за крайния набор от данни, като:

Not Null
Unique
Reference Integrity — референтна цялост (например, customer_id в таблицата orders съответства на id в таблицата customers)
Съответствие с списъка на допустимите стойности

Възможно е добавяне на собствени тестове (custom data tests), като например процент на отклонение в приходите спрямо показателите от преди ден, седмица, месец. Всяко предположение, формулирано под формата на SQL заявка, може да се превърне в тест.

Така могат да се идентифицират нежелани отклонения и грешки в данните в витрините на Хранилището.

Що се отнася до документирането, DBT предоставя механизми за добавяне, версиониране и разпространение на метаданни и коментари на ниво модели и дори атрибути.

Ето как изглежда добавянето на тестове и документация на ниво конфигурационен файл:

 - name: fct_orders
   description: Тази таблица съдържа основна информация за поръчките, както и някои извлечени факти на база плащания
   columns:
    - name: order_id
      tests:
        - unique # проверка за уникалност на стойностите
        - not_null # проверка за наличие на null
      description: Това е уникален идентификатор за поръчка
    - name: customer_id
      description: Външен ключ към таблицата с клиенти
      tests:
        - not_null
        - relationships: # проверка за ссылочна цялост
            to: ref('dim_customers')
            field: customer_id
    - name: order_date
      description: Дата (UTC), на която е направена поръчката
    - name: status
      description: '{{ doc("orders_status") }}'
      tests:
        - accepted_values: # проверка за допустими стойности
            values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']

А ето как изглежда тази документация на генерирания уебсайт:

Маркери и Модули

Целта на DBT не е толкова да бъде набор от SQL скриптове, а да предостави на потребителите мощни и гъвкави инструменти за изграждане на собствени трансформации и разпространение на тези модули.

Маркерите са набори от конструкции и изрази, които могат да бъдат извиквани като функции в моделите. Те позволяват повторна употреба на SQL между модели и проекти в съответствие с инженерния принцип DRY (Не повтаряй себе си).

Пример за маркер:

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

И как се използва:

{% set column_name = 'product' %}
select
 product,
 {{ rename_category(column_name) }} -- извикване на маркера
from my_table

DBT идва с мениджър на пакети (packages), който позволява на потребителите да публикуват и повторно използват отделни модули и маркери.

Това означава възможност за зареждане и използване на библиотеки като:

dbt_utils: работа с Дата/Час, Заместителни ключове, Тестове на схема, Пивот/Деформация и други
Готови шаблони за витрини за такива услуги като Snowplow и Stripe
Библиотеки за определени Хранилища за Данни, например Redshift
Logging — Модул за логване на работата на DBT

С пълния списък на пакетите можете да се запознаете на dbt hub.

Още повече възможности

Тук ще опиша няколко други интересни особености и реализации, които аз и екипът използваме за изграждане на Хранилище на Данни в Wheely.

Разделение на средите DEV — TEST — PROD

Дори в рамките на един и същ кластер DWH (в различни схеми). Например, с помощта на следното изразяване:

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

)

Този код буквално казва: за средите dev, test, ci вземи данни само за последните 3 дни и не повече. Тоест, изпълнението в тези среди ще бъде много по-бързо и ще изисква по-малко ресурси. При стартиране в среда prod условието за филтър ще бъде игнорирано.

Материализация с алтернативно кодиране на колоните

Redshift е колоночна СУБД, позволяваща задаване на алгоритми за компресия на данни за всяка отделна колона. Изборът на оптимални алгоритми може да намали заетия обем на диска с 20-50%.

Макрос redshift.compress_table ще изпълни командата ANALYZE COMPRESSION, ще създаде нова таблица с препоръчителни алгоритми за кодиране на колоните, посочени ключове за сегментация (dist_key) и сортиране (sort_key), ще пренесе данните в нея и при необходимост ще изтрие старата копия.

Сигнатура на макроса:

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

Логиране на стартирания на модели

На всяко изпълнение на модел могат да бъдат добавени хукове (hooks), които ще се изпълняват преди стартиране или веднага след завършване на създаването на модела:

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

Модулът за логиране ще позволи записването на всички необходими метаданни в отделна таблица, по която впоследствие може да се извършва одит и анализ на проблемните области (bottlenecks).

Ето как изглежда таблото на данните за логиране в Looker:

Автоматизация на обслужването на Хранилището

Ако използвате някакви разширения на функционалността на Хранилището, като UDF (User Defined Functions), то версионирането на тези функции, управлението на достъпите и автоматизирането на внедряването на нови версии е много удобно да се осъществява в DBT.

Използваме UDF на Python за изчисляване на хеш стойности, домейни на имейл адреси и декодиране на битови маски (bitmask).

Пример за макрос, който създава UDF на всяка среда за изпълнение (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 %}

В Wheely използваме Amazon Redshift, който е базиран на PostgreSQL. За Redshift е важно редовно да събираме статистики за таблиците и да освобождаваме място на диска — командите ANALYZE и VACUUM, съответно.

С цел това, всяка нощ се изпълняват командите от макроса redshift_maintenance:

{% macro redshift_maintenance() %}
  
   {% set vacuumable_tables=run_query(vacuumable_tables_sql) %}
  
   {% for row in vacuumable_tables %}
       {% set message_prefix=loop.index ~ " от " ~ 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 ~ " Вакуумно почистване на " ~ relation_to_vacuum) }}
           {% do run_query("VACUUM " ~ relation_to_vacuum ~ " BOOST") %}
           {{ dbt_utils.log_info(message_prefix ~ " Анализиране на " ~ 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 ~ " Завършено " ~ relation_to_vacuum ~ " за " ~ total_seconds ~ "s") }}
       {% else %}
           {{ dbt_utils.log_info(message_prefix ~ ' Пропускане на отношение "' ~ row.values() | join ('"."') ~ '" тъй като не съществува') }}
       {% endif %}
  
   {% endfor %}
  
{% endmacro %}

DBT Cloud

Има възможност за използване на DBT като услуга (Managed Service). Включва:

Web IDE за разработка на проекти и модели
Конфигуриране на задачи и график
Лесен и удобен достъп до логовете
Уебсайт с документация на вашия проект
Свързване на CI (Непрекъснато интегриране)

Заключение

Приготвянето и използването на DWH става толкова приятно и полезно, колкото и пиенето на смути. DBT се състои от Jinja, потребителски разширения (модули), компилатор, изпълнител (executor) и мениджър на пакети. Събирайки тези елементи заедно, получавате пълноценно работно пространство за вашето Хранилище на Данни. Днес едва ли може да се намери по-добър начин за управление на трансформациите в DWH.

Представленията, следвани от разработчиците на DBT, се формулират така:

Кодът, а не GUI, е най-добрата абстракция за изразяване на сложна аналитична логика.
Работата с данни трябва да адаптира най-добрите практики за разработка на софтуер.

Най-важната инфраструктура за работа с данни трябва да се контролира от общността на потребителите като софтуер с отворен код.
Не само инструментите за аналитика, но и кодът все по-често ще стане притежание на Open Source общността.

Тези основни убеждения породиха продукт, който днес се използва от над 850 компании и те съставят основата на много интересни разширения, които предстои да бъдат създадени в бъдеще.

За тези, които се интересуват, има видеозапис на открит урок, който проведох преди няколко месеца в рамките на открития урок в OTUS — Data Build Tool за хранилището на Amazon Redshift..

В допълнение към DBT и Хранилищата на Данни, в рамките на курса Data Engineer на платформата OTUS, аз и моите колеги провеждаме занятия по редица други актуални и съвременни теми: