dbt

dbt-selena позволяет использовать dbt для преобразования данных в Selena с применением паттернов моделирования и лучших практик dbt.

dbt-selena GitHub repo.

Experimental feature

Рекомендации по использованию экспериментальных функций

Поддерживаемые функции

Selena >= 3.1	Selena >= 3.4	Функция
✅	✅	Материализация таблиц
✅	✅	Материализация представлений
✅	✅	Материализация материализованных представлений
✅	✅	Инкрементальная материализация
✅	✅	Модель с первичным ключом
✅	✅	Источники
✅	✅	Пользовательские тесты данных
✅	✅	Генерация документации
✅	✅	Партиционирование по выражениям
❌	❌	Kafka
❌	✅	Dynamic Overwrite
*	✅	Submit task
✅	✅	Microbatch (Insert Overwrite)
❌	✅	Microbatch (Dynamic Overwrite)

* Проверьте поддержку submit task для вашей версии, см. SUBMIT TASK

Установка

Установите адаптер Selena DBT с помощью pip:

pip install dbt-selena

Проверка установки

Проверьте установку, проверив версию:

dbt --version

В списке plugins должен отображаться selena.

Конфигурация

Профили

Создайте или обновите profiles.yml с настройками для Selena.

selena_project:
  target: dev
  outputs:
    dev:
      type: selena
      host: your-selena-host.com
      port: 9030
      schema: your_database
      username: your_username
      password: your_password
      catalog: test_catalog

Параметры

type

Описание: Конкретный адаптер для использования, должен быть установлен в selena
Обязательный?: Обязательный
Пример: selena

host

Описание: Имя хоста для подключения
Обязательный?: Обязательный
Пример: 192.168.100.28

port

Описание: Порт для использования
Обязательный?: Обязательный
Пример: 9030

catalog

Описание: Указывает catalog для построения моделей
Обязательный?: Необязательный
Пример: default_catalog

schema

Описание: Указывает schema (база данных в Selena) для построения моделей
Обязательный?: Обязательный
Пример: analytics

username

Описание: Имя пользователя для подключения к серверу
Обязательный?: Обязательный
Пример: dbt_admin

password

Описание: Пароль для аутентификации на сервере
Обязательный?: Обязательный
Пример: correct-horse-battery-staple

version

Описание: Позволяет плагину попытаться использовать совместимую версию selena
Обязательный?: Необязательный
Пример: 3.1.0

use_pure

Описание: установите "true" для использования расширений C
Обязательный?: Необязательный
Пример: true

is_async

Описание: "true" для отправки подходящих задач как etl задач.
Обязательный?: Необязательный
Пример: true

async_query_timeout

Описание: Устанавливает значение query_timeout при отправке задачи в Selena
Обязательный?: Необязательный
Пример: 300

Источники

Создайте или обновите sources.yml

sources:
  - name: your_source
    database: your_sr_catalog
    schema: your_sr_database
    tables:
      - name: your_table

Если catalog не указан в schema, будет использован catalog, определенный в профиле. Используя профиль из предыдущего примера, если catalog не определен, модель будет считать, что источник находится в test_catalog.your_sr_database.

Материализации

Таблица

Базовая конфигурация таблицы

{{ config(
    materialized='table',
    engine='OLAP',
    keys=['id', 'name', 'created_date'],
    table_type='PRIMARY',
    distributed_by=['id'],
    buckets=3,
    partition_by=['created_date'],
    properties=[
        {"replication_num": "1"}
    ]
) }}


SELECT 
    id,
    name,
    email,
    created_date,
    last_modified_date
FROM {{ source('your_source', 'users') }}

Параметры конфигурации

• engine: Движок хранения (по умолчанию: OLAP)
• keys: Столбцы, которые определяют ключ сортировки
• table_type: Тип модели таблицы
  • PRIMARY: Модель с первичным ключом (поддерживает upsert и удаления)
  • DUPLICATE: Модель с дублирующимся ключом (позволяет дублирующиеся строки)
  • UNIQUE: Модель с уникальным ключом (обеспечивает уникальность)
• distributed_by: Столбцы для hash-распределения
• buckets: Количество buckets для распределения данных (оставьте пустым для автоматического bucketing)
• partition_by: Столбцы для партиционирования таблицы
• partition_by_init: Начальные определения партиций
• properties: Дополнительные свойства таблицы Selena

Таблицы во внешних Catalogs

Чтение из внешнего источника в Selena

Этот пример создает материализованную таблицу в Selena, содержащую агрегированные данные из внешнего Hive catalog.

подсказка

Настройте внешний catalog, если он еще не существует:

CREATE EXTERNAL CATALOG `hive_external`
PROPERTIES (
    "hive.metastore.uris"  =  "thrift://127.0.0.1:8087",
    "type"="hive"
);

{{ config(
    materialized='table',
    keys=['product_id', 'order_date'],
    distributed_by=['product_id'],
    partition_by=['order_date']
) }}

-- Агрегирование данных из внешнего Hive catalog в таблицу Selena
SELECT
    h.product_id,
    h.order_date,
    COUNT(*) as order_count,
    SUM(h.amount) as total_amount,
    MAX(h.last_updated) as last_updated
FROM {{ source('hive_external', 'orders') }} h
GROUP BY
    h.product_id,
    h.order_date

Запись во внешний источник

{{
  config(
    materialized='table',
    on_table_exists = 'replace',
    partition_by=['order_date'],
    properties={},
    catalog='external_catalog',
    database='test_db'
  )
}}

SELECT * FROM {{ source('iceberg_external', 'orders') }}

Конфигурация для материализации во внешние catalogs поддерживает меньше опций. Поддерживаются on_table_exists, partition_by и properties. Если catalog и database не установлены, будут использованы значения по умолчанию из профиля.

Инкрементальная

Инкрементальные материализации также поддерживаются в Selena:

{{ config(
    materialized='incremental',
    unique_key='id',
    table_type='PRIMARY',
    keys=['id'],
    distributed_by=['id'],
    incremental_strategy='default'
) }}

SELECT
    id,
    user_id,
    event_name,
    event_timestamp,
    properties
FROM {{ source('raw', 'events') }}

{% if is_incremental() %}
    WHERE event_timestamp > (SELECT MAX(event_timestamp) FROM {{ this }})
{% endif %}

Стратегии инкрементальной загрузки

dbt-selena поддерживает несколько стратегий инкрементальной загрузки:

1. append (по умолчанию): Просто добавляет новые записи без дедупликации
2. insert_overwrite: Перезаписывает партиции таблицы при вставке
3. dynamic_overwrite: Перезаписывает, создает и записывает партиции таблицы

Для получения дополнительной информации о том, какую стратегию перезаписи использовать, см. документацию INSERT.

примечание

В настоящее время инкрементальное слияние не поддерживается.

Устранение неполадок

• Перед использованием внешних catalogs в dbt, вы должны создать их в Selena. Документация по этому вопросу находится здесь.
• Доступ к внешним источникам должен осуществляться с использованием макроса {{ source('external_source_name', 'table_name' }}.
• dbt seed не тестировался для внешних catalogs и в настоящее время не поддерживается.
• Для того чтобы dbt мог создавать модели во внешних базах данных, которые в настоящее время не существуют, расположение моделей должно быть установлено через свойства.
• Внешние модели должны определять расположение, в котором они хранятся. Это расположение будет определено, если целевая база данных существует и устанавливает свойство location. В противном случае расположение необходимо установить.
  • В настоящее время мы поддерживаем создание внешних моделей только в базах данных, которые уже существуют.