Инкрементальное заполнение таблиц фактов хранилища в dbt-модели

блог о bi, №1 в рунете

Data Build Tool (dbt) позволяет поддерживать консистентность и актуальность аналитических данных в хранилище, однако при переходе к масштабируемым проектам возникает задача инкрементального обновления таблиц фактов — с большими объемами событийных данных за длительные периоды. В этой статье мы рассмотрим, как реализовать такую автоматизацию в dbt-моделях с акцентом на контроль периода обновления и предотвращение потери данных.

При переходе с традиционного проекта, где обновление данных выполнялось с помощью отдельных SQL-процедур, на dbt-модели возникает необходимость настроить инкрементальное обновление таблиц фактов. В отличие от справочников, где часто достаточно простой стратегии «merge» по уникальному ключу, таблицы с событиями (фактами) требуют особого подхода из-за объема данных и динамики периодов обновления.

Для таблиц фактов, данные в которые поступают за определённый отчётный период, предпочтительнее использовать стратегию «delete+insert», поскольку она удаляет данные за выбранный период и вставляет актуальные данные заново. Эта стратегия минимизирует возможные рассогласования и упрощает инкрементальное наполнение.

Определение периода обновления через переменные

Обновление данных за период определяется динамически, например, последние полные 2 месяца плюс текущий месяц до даты выполнения обновления. Для задания периода в dbt удобно использовать переменные, передаваемые через шаблонизатор Jinja.

Пример передачи переменных в запуске модели

dbt run --select "fct_currency_rate" --vars '{"date_from": "2025-07-01", "date_to": "2025-09-01"}'

Определение переменных в dbt_project.yml (с указанием значений по умолчанию)

vars: 
   date_from: "2025-07-01"
   date_to: "2025-09-01"

Использование переменных в dbt-модели таблицы фактов

select
  currency_id,
  currency_name,
  currency_rate_rub,
  currency_rate_dt
from {{ ref('stg_1C_ERP_info_register_currency_rates') }}
where currency_rate_dt >= '{{ var("date_from") }}'
  and currency_rate_dt < '{{ var("date_to") }}'

Проверка данных источника и удаление старых данных с помощью hooks

Перед вставкой новых данных в целевую таблицу фактов важно проверить, есть ли данные в источнике за нужный период, чтобы избежать их потери при удалении старых строк из целевой таблицы. Для этого используются hooks - «мини-скрипты» с инструкцией SQL (или списком инструкций SQL), которые автоматически запускаются для выполнения дополнительных действий до (pre-hook) или после (post-hook) какого-то шага в dbt, например:

перед запуском модели (таблицы),
после её успешного создания,
перед или после выполнения тестов,
перед началом или завершением всей сессии.

Пример pre-hook для удаления строк по порциям с проверкой наличия данных

pre_hook = [
    "{% if is_incremental() %}
        if exists (
          select 1 
          from {{ ref('stg_1C_ERP_info_register_currency_rates') }} 
          where currency_rate_date >= '{{ var('date_from') }}' 
            and currency_rate_date < '{{ var('date_to') }}')
        begin
          delete top (100000) from {{this}}
          where currency_rate_dt >= '{{ var('date_from') }}' 
            and currency_rate_dt < '{{ var('date_to') }}'

          while(@@ROWCOUNT > 0)
          begin
            delete top (100000) from {{this}}
            where currency_rate_dt >= '{{ var('date_from') }}' 
              and currency_rate_dt < '{{ var('date_to') }}'
          end
        end
      {% endif %}"
]

В данном pre-hook мы проверяем:

с помощью «if is incremental()» убеждаемся, что целевая таблица существует, её материализация – incremental, а запуск dbt-модели целевой таблицы осуществляется без флага –full-refresh (который пересоздает dbt-модель заново)
условие «if exists» выполняется при наличии данных в таблице-источнике stg_1C_ERP_info_register_currency_rates за нужный период
если данные в таблице-источнике существуют – удаляем из целевой таблицы (которая указывается как {{this}} – текущая модель, в которой написан pre-hook) данные за период обновления в цикле по 100.000 строк

После выполнения pre-hook с помощью запроса добавляем в целевую таблицу данные за период обновления, устанавливаемый с помощью переменных, а также после создания (или полного перестроения с помощью –full-refresh) таблицы с помощью post-hook пишем скрипт команды SQL для создания кластеризованного индекса по дате, что оптимизирует производительность запросов к модели с фильтрацией по дате.

Пример полной dbt-модели с инкрементальным обновлением фактов

{{
  config(
    alias="fct_currency_rate",
    schema="1c_erp",
    materialized="incremental",
    incremental_strategy="delete+insert",
    pre_hook = [
      "{% if is_incremental() %}
          if exists (
            select *
            from {{ ref('stg_1C_ERP_info_register_currency_rates') }}
            where currency_rate_date >= '{{ var('date_from') }}'
              and currency_rate_date < '{{ var('date_to') }}')
          begin
            delete top (100000) from {{this}}
            where currency_rate_dt >= '{{ var('date_from') }}' 
              and currency_rate_dt < '{{ var('date_to') }}'

            while(@@ROWCOUNT > 0)
            begin
              delete top (100000) from {{this}}
              where currency_rate_dt >= '{{ var('date_from') }}' 
                and currency_rate_dt < '{{ var('date_to') }}'
            end
          end
        {% endif %}"
    ],
    post_hook  = [
      "{% if not is_incremental() %}
        create clustered index ix_cl_{{ this.name }}_currency_rate_dt on {{ this }} (currency_rate_dt);
      {% endif %}"
    ]
  )
}}

select
  currency_id,
  currency_name,
  currency_rate_rub,
  currency_rate_dt
from {{ ref('stg_1C_ERP_info_register_currency_rates') }}
where currency_rate_dt >= '{{ var('date_from') }}'
  and currency_rate_dt < '{{ var('date_to') }}'

Вывод

На основании вышеизложенного можно сделать вывод о том, что dbt обеспечивает удобную и надёжную платформу для построения и управления аналитическими моделями, а для инкрементального обновления таблиц фактов с динамическими периодами загрузки данных оптимальным решением является материализация с использованием стратегии «delete+insert», обеспечивающей чистоту и соответствие данных.

Использование переменных Jinja для определения периода обновления при запуске dbt-моделей позволяет гибко управлять загрузками без постоянного изменения SQL-кода, а pre-hook и post-hook дают дополнительный контроль за целостностью данных и общей производительностью.

Таким образом, организовать отказоустойчивое и консистентное обновление фактов в хранилище с помощью dbt-моделей возможно при правильной настройке процессов и применении инструментов самого фреймворка. Это повышает качество и актуальность аналитики при минимальных операционных затратах.

ВЕРНУТЬСЯ К БЛОГУ