Использование DBT для построения архитектуры Medallion Lakehouse (Azure Databricks + Delta + DBT)
Инструмент построения данных (DBT) наделал очень много шума. Он набирает все большую и большую популярность в сообществе разработчиков данных. Я видел несколько проектов, где DBT уже выбран в качестве главного инструмента. Но что же это такое? Как Вы можете использовать его в своем ландшафте данных? Что он делает и как работает?
Эта статья посвящена ответам на все эти вопросы. Я продемонстрирую, как работает DBT на примере архитектуры DataBricks. Предупреждаю! Чтение займет много времени…
Прежде чем мы начнем что-либо настраивать, давайте изучим DBT.
Что такое dbt?
DBT - это инструмент трансформации в процессе ELT. Это инструмент командной строки с открытым исходным кодом, написанный на языке Python. DBT фокусируется на "T" в процессе ELT (Extract, Transform and Load), поэтому он не извлекает и не загружает данные, а только преобразует их.
DBT поставляется в двух вариантах: "DBT core" – cli - версия с открытым исходным кодом и платная версия "DBT cloud". В этой статье я буду использовать бесплатную версию cli. Мы будем использовать ее после запуска конвейера с помощью Фабрики данных Azure (ADF) и займемся преобразованием данных с помощью DBT.
Сила DBT заключается в том, что он осуществляет преобразования с помощью шаблонов. Синтаксис похож на операторы SELECT в SQL. Кроме того, весь поток строится в виде прямого ациклического графа (DAG), что позволяет визуализировать данные, включая историю данных.
DBT отличается от других инструментов тем, что основан на шаблонах и управляется драйвером CLI. Таким образом, вместо визуального проектирования ETL Вы настраиваете преобразования с помощью шаблонов SQLBoiler. Преимущество такого подхода заключается в том, что Вы не сильно зависите от базовой базы данных. Это означает то, что Вы достаточно легко можете переходить от одного поставщика баз данных к другому. Кроме того, Вам не придется изучать множество языков баз данных - DBT автоматически транспилирует или генерирует код, необходимый для преобразования данных.
Пример
В качестве примера мы будем использовать базу данных Azure SQL, настроенную на работу с образцами данных - AdventureWorks. Она будет играть роль источника, из которого мы будем получать данные. Для хранения и постепенного улучшения структуры наших данных мы будем использовать такие сервисы, как Azure Data Lake Services (ADLS), Azure Data Factory, Azure DataBricks и DBT. Конечной целью является создание простой и удобной модели данных, полностью готовой к дальнейшему использованию.
Предварительные условия
Прежде чем приступать к работе, неплохо бы иметь несколько установленных программ. У Вас должна быть установлена последняя версия Python. Для Azure создайте группу ресурсов, содержащую следующие службы:
- Azure Databricks, используя руководство по быстрому запуску (QuickStart);
- ADLS gen2, используя учебное пособие по созданию аккаунта Data Lake Storage (create-data-lake-storage-account);
- Azure SQL Server, включая образец БД AdventureWorks;
- Azure Data Factory, используя руководство по быстрому запуску (QuickStart);
- Azure Key Vault, используя руководство по быстрому запуску (QuickStart) .
Рекомендуется развернуть эти службы в порядке, указанном выше. При этом Ваша учетная запись DataBricks автоматически получит доступ к Key Vault.
Конфигурация — Создание Контейнера
После развертывания всех служб перейдите в только что развернутую учетную запись хранилища. Создайте три контейнера: бронзовый, серебряный и золотой.
Немного информации о структуре.
Бронзовый контейнер будет использоваться для захвата всех необработанных данных. Мы будем использовать файлы Parquet, так как версионность не требуется. Для добавления новых загруженных данных в этот контейнер мы будем использовать схему YYYYMMDD.
Серебряный контейнер будет использоваться для слегка преобразованных и стандартизированных данных. Формат файлов - дельта. Для объединения различных наборов данных будет использоваться DBT.
Наконец, золото, которое будет задействовано для окончательного интегрирования данных. И снова мы будем использовать DBT для объединения различных наборов данных, а формат файла также будет дельта.
Фабрика данных Azure— новый конвейер
Далее мы будем использовать ADF для построения нашего первого конвейера для ввода данных в бронзовый слой. Здесь данные будут разделены по времени. После загрузки данных мы будем использовать тот же конвейер для вызова Databricks для настройки всех схем (внешних таблиц) для чтения данных.
Перед созданием нового конвейера убедитесь, что Ваши службы ADLS и Azure SQL добавлены как связанные службы: https://learn.microsoft.com/en-us/azure/data-factory/concepts-linked-services?tabs=data-factory
После настройки связанных служб начните создание нового конвейера в ADF. Конечный результат показан на рисунке ниже. Для конвейера Вам понадобятся: Lookup, ForEach, CopyTables и Notebook.
Далее мы обсудим, как подключить данные из базы данных AdventureWorks. Для создания действий Lookup и ForEach я рекомендую посмотреть это видео: https://www.youtube.com/watch?v=KsO2FHQdILs. Для всего потока нам необходимы три набора данных: TableList, SQLTable, ParquetTable.
Первый набор данных использует службу Linked Service для Azure SQL. Он предназначен для составления списка всех таблиц. Можно, например, назвать ее TableList. Никакой особой настройки для этой связанной службы не требуется:
Второй набор данных использует ту же самую службу Linked Service для Azure SQL. Этот набор данных предназначен для копирования всех таблиц. Вы можете назвать его SQLTable. Для динамической настройки схемы и имен таблиц используйте следующие два параметра:
@dataset().SchemaName @dataset().TableName
Также убедитесь в том, что параметры установлены соответствующим образом:
После настройки наборов данных Azure SQL DataSets вернитесь к конвейеру. Перетащите Lookup, выберите DataSet для получения списка всех таблиц. Мы будем использовать запрос для получения всех нужных нам таблиц. Также убедитесь в том, что Вы сняли флажок только для первой строки.
SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE' AND TABLE_SCHEMA = 'SalesLT'
Убедитесь, что все работает должным образом, нажав кнопку Preview Data. После этого перетащите ForEach. Для итерации информации об объектах sys, возвращаемой SQL Server, добавьте в Items следующую информацию:
@activity('List Tables').output.value
Важно, чтобы входное имя задания точно совпадало с именем предыдущего задания. Так, в данном примере оно должно указывать на задание List Tables. При правильной настройке Ваши имена схем и таблиц будут извлечены из результатов запроса и переданы в качестве аргументов в цикл ForEach.
Далее откройте цикл ForEach и перетащите в него задание CopyTables. Выберите набор данных SQLTable, содержащий входные параметры. Используйте TableName и SchemaName. Добавьте следующую информацию:
@item().table_name @item().table_schema
Перейдите к Sink. Добавьте новый набор данных DataSet. Как вариант, этот набор данных может называться ParquetTable. В качестве формата файла выберите Parquet. В диалоговом окне Parameters добавьте два свойства: FileName и FilePath.
Перейдите к Connection tab. Выберите Browse. Перейдите к бронзовому контейнеру. В данном примере мы не будем использовать вложенные папки приложений. Расширьте путь к файлу двумя параметрами:
@dataset().FilePath @dataset().FileName
Вернитесь к конвейеру и свойствам Sink. Для FileName и FilePath используйте текст, приведенный ниже:
@concat(item().table_schema,'.',item().table_name,'.parquet') @formatDateTime(utcnow(), 'yyyyMMdd')
Как только все настроено, нажмите кнопку Add trigger и наблюдайте за процессом. Если все работает как надо, Вы увидите папку в бронзовом контейнере. В папке будет указано сегодняшнее число.
В каждой папке хранится набор таблиц с файлами Parquet. На следующих этапах мы достроим конвейер, добавив схемы в DataBricks. Таким образом, при каждом запуске конвейера схемы будут обновляться и указывать на последние данные. Давайте перейдем к DataBricks, чтобы продолжить наше «путешествие».
Конфигурация —подключение DataBricks
Для настройки DataBricks я рекомендую использовать DataBricks CLI. Убедитесь, что у Вас установлен Python. Чтобы установить CLI, выполните следующую команду от имени администратора:
pip install databricks-cli
Для доступа к DataBricks я буду использовать Access Token. Вы можете управлять им в настройках пользователя. Итак, щелкните справа вверху на имя пользователя, выберите User settings. Нажмите на кнопку Generate new token. Добавьте имя и закрепите токен. Вам придется использовать его несколько раз, поэтому я рекомендую хранить его в безопасном месте.
Вернитесь в командную строку и выполните следующую команду:
databricks configure --token
Вам будет предложено ответить на два вопроса. Для хоста используйте адрес https. В моем случае: https://adb-7945672859019303.3.azuredatabricks.net/. После копирования и вставки полного HTTP-адреса Вам будет предложено вставить токен. После этого должно произойти подключение. Вы можете проверить его, используя следующую команду:
databricks fs ls
Если команда успешно выполнена, Вы увидите имена папок, например databricks-results и user.
Конфигурация — DataBricks и KeyVault
Далее я рекомендую создать соединение между DataBricks и Key Vault. Key Vault - это служба, в которой могут храниться учетные данные. Key Vault необходим, когда DataBricks хочет получить доступ к Вашему озеру данных.
Чтобы установить соединение между Key Vault и DataBricks, Вам нужно создать область секретов. Это можно сделать, добавив #secrets/createScope в конце url Вашего рабочего пространства, в моем случае: https://adb-7945672859019303.3.azuredatabricks.net/?o=7945672859019303#secrets/createScope.
Введите имя области, а также добавьте DNS-имя и идентификатор ресурса для Key Vault. Их можно найти на портале Azure Portal справа. Итак, выберите свойства и скопируйте URI хранилища и идентификатор ресурса.
После завершения работы убедитесь, что область секретов указана в списке. Для этого используйте командную строку. Введите следующее:
databricks secrets list-scopes
Продолжаем настраивать Key Vault: перейдите в раздел Key Vault Secrets и добавьте новый секрет для доступа к учетной записи хранилища. Создайте новый секрет и введите имя, например blobAccountKey.
Значение секрета должно быть ключом доступа из Вашей учетной записи хранилища. Его можно найти в учетной записи хранилища в разделе Access Keys.
После создания секрета убедитесь, что Ваша служба DataBricks имеет права на список и чтение Ваших секретов. Это можно настроить в разделе Access Policies.
Чтобы убедиться, что все работает должным образом, создайте новый блокнот DataBricks и введите в него следующую информацию:
abcd = dbutils.secrets.get('dbtScope','blobAccountKey')
print(abcd)
Запустите блокнот. Ответ REDACTED означает, что Ваше рабочее пространство Databricks может получить доступ к Вашей учетной записи хранилища с помощью Key Vault.
Далее я рекомендую добавить точки монтирования к контейнерам хранения. Создайте еще один блокнот и выполните следующий код для добавления точек монтирования к бронзовому, серебряному и золотому контейнерам:
#mount bronze
dbutils.fs.mount(
source='wasbs://bronze@dbtdemo.blob.core.windows.net/',
mount_point = '/mnt/bronze',
extra_configs = {'fs.azure.account.key.dbtdemo.blob.core.windows.net': dbutils.secrets.get('dbtScope','blobAccountKey')}
)
#mount silver
dbutils.fs.mount(
source='wasbs://silver@dbtdemo.blob.core.windows.net/',
mount_point = '/mnt/silver',
extra_configs = {'fs.azure.account.key.dbtdemo.blob.core.windows.net': dbutils.secrets.get('dbtScope','blobAccountKey')}
)
#mount gold
dbutils.fs.mount(
source='wasbs://gold@dbtdemo.blob.core.windows.net/',
mount_point = '/mnt/gold',
extra_configs = {'fs.azure.account.key.dbtdemo.blob.core.windows.net': dbutils.secrets.get('dbtScope','blobAccountKey')}
)
dbutils.fs.ls("/mnt/bronze")
dbutils.fs.ls("/mnt/silver")
dbutils.fs.ls("/mnt/gold")
Обратите внимание, что расположение ADLS будет выглядеть несколько иначе, поэтому обновите приведенный выше код, используя имя Вашего блоба.
Фабрика данных Azure— схемы DataBricks
Когда все будет готово, давайте вернемся к Фабрике данных Azure, чтобы расширить наш конвейер данных. Создайте новую связанную службу. Я использую тот же персональный маркер доступа, но Вы можете настроить свою связанную службу по-другому, следуя руководству на сайте MSLearn.
Убедитесь, что соединение работает, протестировав его с помощью кнопки справа. Далее вернитесь к Вашему конвейеру. В активности ForEach добавьте новую активность DataBricks Notebook, но перед этим нам нужно создать скрипт в самой DataBricks. Предлагаемый мной скрипт имеет следующий код:
#fetch parameters from Azure Data Factory
table_schema=dbutils.widgets.get("table_schema")
table_name=dbutils.widgets.get("table_name")
filePath=dbutils.widgets.get("filePath")
#create database
spark.sql(f'create database if not exists {table_schema}')
#create new external table using latest datetime location
ddl_query = """CREATE OR REPLACE TABLE """+table_schema+"""."""+table_name+"""
USING PARQUET
LOCATION '/mnt/bronze/"""+filePath+"""/"""+table_schema+"""."""+table_name+""".parquet'
"""
#execute query
spark.sql(ddl_query)
Скрипт получает параметры из ADF и использует их для удаления всех существующих таблиц и создания новых, используя последнее расположение YYYYMMDD. После создания скрипта вернитесь к конвейеру ADF и блокноту.
В разделе Настройки в блокноте выберите только что созданный скрипт. Добавьте три базовых параметра table_schema, table_name и filePath. Вы можете использовать текст, приведенный ниже:
@item().table_schema @item().table_name @formatDateTime(utcnow(), 'yyyyMMdd')
После добавления этого шага снова нажмите кнопку Add trigger. После этого перейдите в DataBricks и просмотрите раздел данных. Если все работает так, как было запланировано, Ваше новое имя таблицы должно появиться в базе данных saleslt.
Выполнив все действия, описанные выше, мы заложили основу для дальнейшей работы. Каждый день при запуске конвейера будут создаваться новые папки, туда будут копироваться необработанные данные, а таблицы DataBricks будут удаляться и создаваться заново. Теперь давайте изучим, что же DBT может нам предложить.
DBT — установка и конфигурация
Теперь пришло время установить и настроить DBT. Для этого я рекомендую сначала создать среду для DBT. Это можно сделать, выполнив следующие команды:
python3 -m venv dbt-env
Затем активируйте среду:
dbt-env\Scripts\activate
Далее используйте командную строку и введите следующие команды для установки dbt:
pip install dbt-databricks
После установки убедитесь, что DBT работает. Введите следующее:
dbt --version
DBT — Подключение DBT к DataBricks
Теперь, когда DBT установлен, Вы можете создать новый проект DBT. Это можно сделать с помощью команды dbt init. Для того, чтобы хранить свой код на GitHub, перейдите в локально установленную директорию GitHub и создайте новый проект с помощью команды init. Прежде чем сделать это, убедитесь в том, что Вы скопировали необходимую информацию о проекте из рабочей области DataBricks. Перейдите к настройкам кластера и скопируйте свой токен, имя хоста сервера и HTTP. Эта информация нужна Вам для запуска проекта.
Затем создайте новый проект с помощью следующей команды:
dbt init
Ответьте на все вопросы, выбрав Databricks, добавив свой хост, https_path и default в качестве имени схемы. После того как Вы все настроили, проверьте соединение, перейдя в папку с только что созданным проектом. Далее запустите команду dbt debug:
Как видите, все работает. Давайте начнем строить Вашу первую логику преобразования! На случай, если что-то пошло не так или Вы хотите изменить токен или конфигурацию, имейте в виду, что файл конфигурации DBT хранится в папке Вашего профиля. В моем случае здесь: C:\Users\pstrengholt\.dbt\profiles.yml
DBT — построение медленно меняющихся имерений
Открыв только что созданный файл проекта, Вы увидите несколько папок. Папка Models предназначена для построения логики преобразований. Если Вы откроете эту папку, то увидите два файла: my_first_dbt_model.sql и my_second_dbt_model.sql. Также Вы увидите файл schema.yml, в котором описана структура этих данных. Я рекомендую удалить всю папку с примерами, потому что в этом уроке мы все начнем с нуля. После этого я также рекомендую обновить файл dbt_project.yml, который находится в корне папки Вашего проекта. Очистите секцию примеров в разделе Models.
Прежде чем моделировать или историзировать данные, сначала нужно определить существующие таблицы, которые находятся в DataBricks. Создайте в папке Models папку Staging. В ней создайте файл Sources.yml. Добавьте в него следующий код:
version: 2
sources:
- name: saleslt
schema: saleslt
description: This is the adventureworks database loaded into bronze
tables:
- name: address
- name: customer
- name: customeraddress
В этой демонстрационной версии мы будем использовать только три таблицы: address, customer и customeraddress. Добавленные источники указывают на данные, которые уже находятся в DataBricks. Помните, что мы создали бронзовое представление, которое указывает на самые последние файлы parquet в озере.
Помимо моделей Вы также видите папку snapshots . Файлы, создаваемые в этой папке, отлично подходят для историзации данных. Внутри этой папки создайте новый файл под названием: customer.sql:
customer.sql
{% snapshot customer_snapshot %}
{{
config(
file_format = "delta",
location_root = "/mnt/silver/customer",
target_schema='snapshots',
invalidate_hard_deletes=True,
unique_key='CustomerId',
strategy='check',
check_cols='all'
)
}}
with source_data as (
select
CustomerId,
NameStyle,
Title,
FirstName,
MiddleName,
LastName,
Suffix,
CompanyName,
SalesPerson,
EmailAddress,
Phone,
PasswordHash,
PasswordSalt
from {{ source('saleslt', 'customer') }}
)
select *
from source_data
{% endsnapshot %}
Если Вы изучите файл, то увидите, что мы начинаем с определения секции моментальных снимков. Мы указываем параметры для использования дельта-формата и место хранения данных. В нашем случае это /mnt/silver/customer. Кроме того, мы предоставляем информацию о том, как должна быть определена дельта. После этого мы выбираем все необходимые данные из данных sales.customer. Среда должна выглядеть так, как показано на скриншоте ниже:
После того как Вы создали исходные тексты и скрипт моментального снимка, пришло время вернуться в терминал. Введите следующее:
dbt snapshot
Давайте перейдем к DataBricks и посмотрим, что же у нас получилось. Если Вы заглянете в раздел данных, то увидите вновь созданную базу данных, которая называется "snapshot". Если все прошло успешно, Вы также увидите новую таблицу под названием: customer_snapshot.
Прокрутите страницу вниз или до конца вправо. Вы увидите новые столбцы dbt_valid_from и dbt_valid_to. Эти столбцы используются для отслеживания вновь созданных данных. Каждый раз, когда запись изменяется, она добавляется в таблицу. В столбце dbt_valid_to для старых записей будет указана дата, до которой записи действительны. Все новые записи будут иметь пустое значение dbt_valid_to.
Я рекомендую Вам также заглянуть в серебряный контейнер в Вашем озере данных. Вы увидите недавно созданную папку customer, в которой хранятся файлы в формате дельта.
Чтобы завершить этап историзации, добавьте еще два файла:
После добавления всех файлов снова запустите процесс моментального снимка. Теперь у Вас должно быть три новых таблицы.
DBT — Построение логики интеграции для золотого слоя
Следующий шаг - это использование серебряного слоя в качестве исходных данных для золотого слоя. В золотом слое данные обычно более интегрированы и удобны для использования. Итак, я хочу создать денормализованную таблицу, объединив все три таблицы из серебряного слоя.
Создайте новую папку в каталоге Models под названием Marts. В ней создайте два новых файла. Первый под названием: dim_customers.sql
{{
config(
materialized = "table",
file_format = "delta",
location_root = "/mnt/gold/customers"
)
}}
with address_snapshot as (
select
AddressID,
AddressLine1,
AddressLine2,
City,
StateProvince,
CountryRegion,
PostalCode
from {{ ref('address_snapshot') }} where dbt_valid_to is null
)
, customeraddress_snapshot as (
select
CustomerId,
AddressId,
AddressType
from {{ref('customeraddress_snapshot')}} where dbt_valid_to is null
)
, customer_snapshot as (
select
CustomerId,
-- Adopted function concat() to concatenate first, middle and lastnames
concat(ifnull(FirstName,' '),' ',ifnull(MiddleName,' '),' ',ifnull(LastName,' ')) as FullName
from {{ref('customer_snapshot')}} where dbt_valid_to is null
)
, transformed as (
select
row_number() over (order by customer_snapshot.customerid) as customer_sk, -- auto-incremental surrogate key
customer_snapshot.CustomerId,
customer_snapshot.fullname,
customeraddress_snapshot.AddressID,
customeraddress_snapshot.AddressType,
address_snapshot.AddressLine1,
address_snapshot.City,
address_snapshot.StateProvince,
address_snapshot.CountryRegion,
address_snapshot.PostalCode
from customer_snapshot
inner join customeraddress_snapshot on customer_snapshot.CustomerId = customeraddress_snapshot.CustomerId
inner join address_snapshot on customeraddress_snapshot.AddressID = address_snapshot.AddressID
)
select *
from transformed
Если Вы посмотрите на новый файл, то увидите, что мы снова используем формат дельта. На этот раз мы выводим данные в наш золотой контейнер. Вы также можете видеть, что мы применяем простую бизнес-логику. Например, я объединяю три столбца для создания FullName.
И второй файл под названием: dim_customers.yml
version: 2
models:
- name: dim_customers
columns:
- name: customer_sk
description: The surrogate key of the customer
tests:
- unique
- not_null
- name: customerid
description: The natural key of the customer
tests:
- not_null
- unique
- name: fullname
description: The customer name. Adopted as customer_fullname when person name is not null.
- name: AddressId
tests:
- not_null
- name: AddressType
- name: AddressLine1
- name: City
- name: StateProvince
- name: CountryRegion
- name: PostalCode
В этом файле содержится информация о схеме нового интегрированного набора данных. В нем содержатся данные о первичном ключе и выполняются некоторые тесты на уникальность данных.
Все готово, теперь давайте выясним, можно ли правильно обработать золотой слой. Вернитесь в командную строку. На этот раз выполните команду:
dbt run
После завершения процесса запуска вернитесь в рабочую область DataBricks и загляните в раздел data. Вы найдете только что созданную таблицу под названием: dim_customer. Эта таблица, как Вы можете видеть ниже, содержит все интегрированные данные.
Отлично! Вы успешно скопировали данные из Azure SQL и обработали их, пройдя несколько этапов.
DBT — Документация и история
Последняя функция, которую я хочу показать, - это генерация документации. Вернитесь в терминал и введите следующие команды:
dbt docs generate dbt docs serve
Для того, чтобы увидеть всю документацию по Вашему проекту, дождитесь открытия браузера или вручную перейдите на страницу http://localhost:8080. Нажмите на зеленую кнопку в правой части экрана. Ниже приведен пример того, что у Вас должно получиться:
Заключение
Вы потратили достаточно много времени на чтение данной статьи и узнали, как легко построить архитектуру DataBricks, используя дизайн Lakehouse. Вы также узнали о получении и разделении данных с помощью структуры времени даты и увидели, как можно быстро создать серебряный слой с помощью операторов select и моментального снимка. Кроме того, Вы научились объединять и интегрировать данные. При дальнейшем масштабировании предлагаю следовать некоторым советам:
- Рекомендуется использовать разные базы данных Databricks и расположение папок ADLS для приложений. Пока мы создали все в корне, но при добавлении новых источников рекомендуется создать определенную структуру;
- Рассмотрите возможность версионирования дельта-файлов. Это позволит Вам быстро откатиться назад в случае обработки поврежденных или неверных данных;
- Рекомендуется развернуть сервер запуска и сборки. В производственной среде весь процесс должен быть оркестрован правильно. Рассмотрите возможность использования виртуальных машин или DBT Cloud. Для добавления DBT в Ваш конвейер я рекомендую ознакомиться со следующей информацией: https://medium.com/@guangx/run-dbt-in-azure-data-factory-a-clean-solution-for-azure-cloud-edddf0c85849
- Настоятельно рекомендуется использовать защищенные частные конечные точки. Более подробную информацию об этом можно найти здесь: https://learn.microsoft.com/en-us/azure/databricks/administration-guide/cloud-configurations/azure/private-link
- Если Вы хотите загрузить все скрипты, воспользуйтесь следующим репозиторием github: https://github.com/pietheinstrengholt/dbt-databricks-adventureworks
