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
