Data Vault за один вечер

Data Vault за один вечер

Решила я тут погрузиться в дата волт на практике. Нашла интересный проект — AutomateDV (бывший dbtvault)

У них есть демо-проект, в котором создаются разные сущности, несколько видосов с объяснениями и небольшая, но очень детальная дока

💻 Поставила себе по инструкции dbt-core, dbt-postgres:

python -m pip install dbt-core dbt-postgres

Подключила AutomateDV — это просто пакет в dbt-проекте, задается в packages.yml:

dbt deps

Подняла постгрю в докере, настроила коннекшены, создала проект:

dbt init

Сгенерила данные из датасета TPC-H и положила их в volume:

pip install tpchgen-cli

mkdir data

tpchgen-cli -s 1 -f csv -o ./data

Создала таблицы и загрузила в них данные:

COPY region FROM '/var/lib/postgresql/data/data/region.csv' DELIMITER ',' CSV HEADER;

🐾 Поначалу я смотрела видосы и делала все ручками. Потом забила и чисто запускала модельки, смотрела, что получается. Фреймворк автоматизирует создание дата волта. Во-первых, для стейджинга можно использовать такие поля:

derived_columns - новые поля

hashed_columns - поля для хэширования

Тут мы добавили поле как алиас другого, поле с константной строкой, сделали хэш из 2х первичный ключей:

{%- set yaml_metadata -%}

source_model: 'raw_orders'

derived_columns:

CUSTOMER_KEY: 'CUSTOMERKEY'

RECORD_SOURCE: '!TPCH-ORDERS'

hashed_columns:

ORDER_CUSTOMER_PK:

- 'CUSTOMER_KEY'

- 'ORDERKEY'

...

Во-вторых, создание самих сущностей происходит с помощью удобных макросов:

automate_dv.stage()

automate_dv.hub()

automate_dv.sat()

automate_dv.link()

automate_dv.pit()

...

Все модельки можно позапускать, а потом сгенерить доку по проекту:

dbt run -s +v_stg_orders

dbt run -s hub_customer

dbt docs generate

dbt docs serve

😅 Но саааамое прикольное — это после компиляции моделек зайти в папку target/ и посмотреть, что там нагенерилось. И еще сходить в исходный код самого пакета в dbt_packages/

А нагенерилось там вот такое:

—хэширование нескольких полей

DECODE(MD5(NULLIF(CONCAT_WS('||',

COALESCE(NULLIF(UPPER(TRIM(CAST(CUSTOMER_KEY AS VARCHAR))), ''), '^^'),

COALESCE(NULLIF(UPPER(TRIM(CAST(ORDERKEY AS VARCHAR))), ''), '^^')

), '^^||^^')), 'hex') AS ORDER_CUSTOMER_PK,

—отбор по rn для инсерта в линки

ROW_NUMBER() OVER(

PARTITION BY rr.ORDER_CUSTOMER_PK

ORDER BY rr.LOAD_DATE

) AS row_number

—отбор уникальных значений в сателлит по хешу

WHERE b.INVENTORY_HASHDIFF != b.prev_hashdiff

В общем, интересно поковыряться, как они придумали, и вдохновиться, если у вас сложнее/что-то забыли учесть/еще не реализовали

@data_engineerette