dbt incremental models на собеседовании Data Engineer

Зачем спрашивают на собесе DE

dbt — стандарт T в ELT. Incremental models — главный приём для больших таблиц. На собесе DE: «как написать incremental», «выбор стратегии», «как обрабатывать late-arriving data».

Зачем incremental

Полный refresh большой fact-таблицы — дорого. На 1ТБ и ежедневном rebuild — невозможно.

Incremental — обрабатывает только новые / изменённые строки с момента последнего запуска.

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge'
) }}

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

{% if is_incremental() %}
  WHERE updated_at > (SELECT COALESCE(MAX(updated_at), '1900-01-01') FROM {{ this }})
{% endif %}

Первый запуск — полный SELECT. Последующие — только новые updated_at.

Стратегии

merge — стандарт. Использует MERGE / UPSERT в БД с поддержкой (Postgres 15+, Snowflake, BigQuery, Spark).

MERGE INTO target USING source ON target.unique_key = source.unique_key
WHEN MATCHED → UPDATE
WHEN NOT MATCHED → INSERT

Требует unique_key.

append — просто INSERT новых записей. Без update. Без проверки дубликатов.

{{ config(materialized='incremental', incremental_strategy='append') }}

Подходит для append-only данных (логи, события).

insert_overwrite — переписать партицию целиком.

{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={'field': 'event_date', 'data_type': 'DATE'}
) }}

Хорошо для time-partitioned таблиц на BigQuery / Spark / Databricks.

delete+insert — DELETE по unique_key, потом INSERT. Для адаптеров без MERGE.

is_incremental() и пример модели

{{ this }} — ссылка на саму модель (целевую таблицу).

is_incremental() — true, если выполняется как incremental (а не первый run).

{{ config(
    materialized='incremental',
    unique_key=['user_id', 'event_date'],
    on_schema_change='append_new_columns'
) }}

WITH source AS (
  SELECT
    user_id,
    DATE_TRUNC('day', event_at) AS event_date,
    COUNT(*) AS events,
    SUM(amount) AS revenue
  FROM {{ source('raw', 'events') }}

  {% if is_incremental() %}
    WHERE event_at >= (SELECT MAX(event_date) FROM {{ this }})
  {% endif %}

  GROUP BY 1, 2
)

SELECT * FROM source

on_schema_change — что делать, если в источнике появилась новая колонка: ignore, append_new_columns, sync_all_columns, fail.

Late-arriving data

Проблема. Некоторые события приходят с опозданием. Если фильтруем WHERE event_at > MAX(event_at), опоздавшие выпадают.

Решения:

1. Lookback window.

{% if is_incremental() %}
  WHERE event_at >= (SELECT MAX(event_date) FROM {{ this }}) - INTERVAL '7 days'
{% endif %}

Перепроверяем последние 7 дней — поздние события подхватываются.

2. Updated_at вместо event_at.

WHERE updated_at > (SELECT MAX(updated_at_runtime) FROM {{ this }})

Источник должен подтверждать updated_at. Тогда любые изменения — пойманы.

3. CDC.

С Debezium / Kafka — приходят все изменения, ничего не пропускаем.

Важно: при lookback нужна стратегия merge (или delete+insert), иначе будут дубликаты.

Full refresh и snapshots

Full refresh. Полный пересчёт модели. Запускается через dbt run --full-refresh.

Используется когда:

• Изменили логику модели.
• Сломалась схема, нужно пересчитать.
• Plain incremental накопил inconsistency.

Snapshot — отдельная стратегия для SCD2.

{% snapshot dim_users_history %}
{{ config(
    target_schema='snapshots',
    unique_key='user_id',
    strategy='TIMESTAMP',
    updated_at='updated_at'
) }}

SELECT * FROM {{ source('raw', 'users') }}

{% endsnapshot %}

Snapshot хранит историю изменений с dbt_valid_from / dbt_valid_to.

Частые ошибки

Incremental без unique_key для merge. Дубликаты при повторных запусках.

Использовать event_at вместо updated_at. Late events не подхватятся.

Не делать lookback. Пропуски late events.

on_schema_change='ignore' в production. Новые колонки источника тихо теряются.

Запускать dbt run в нескольких экземплярах. Race condition при INSERT — лочить через advisory lock или ставить серийно.

Полный refresh ежедневно. Теряем смысл incremental. Refresh — редко (раз в неделю / месяц).

Тестировать только на маленьких данных. Production-логика incremental может ломаться на edge cases (timezone, NULL в updated_at).

Связанные темы

• MERGE и UPSERT для DE
• SCD типы для DE
• dbt на собесе DE
• Идемпотентность пайплайна для DE
• Подготовка к собесу Data Engineer

FAQ

merge или delete+insert?

merge быстрее, если БД поддерживает (Postgres 15+, Snowflake, BigQuery). delete+insert — fallback для старых движков.

Insert_overwrite в BigQuery — это что?

Перезаписывает целые партиции. Идиома: фильтр в WHERE по партициям, dbt автоматически делает MERGE с overwrite этих партиций.

Можно ли инкрементально джойнить?

Да, но осторожно. Если правая таблица обновилась — инкремент левой не подхватит изменений. Часто проще full refresh для join-моделей.

Это официальная информация?

Нет. Статья основана на документации dbt 1.7+ и материалах dbt Labs.

Тренируйте Data Engineering — откройте тренажёр с 1500+ вопросами для собесов.