Инкрементальные подходы для обработки данных почти в реальном времени

В этом разделе рассматриваются три основных инкрементальных подхода, позволяющих поддерживать актуальность данных почти в реальном времени:

Инкрементальный MERGE из таблиц только для добавления
CDC с использованием Snowflake Streams
Микропакетная обработка больших таблиц временных рядов

Подходы, специфичные для Snowflake

Некоторые подходы в этом руководстве используют возможности, специфичные для Snowflake. В других хранилищах есть аналогичные возможности с другими реализациями. Документацию для конкретных адаптеров см. в разделе дополнительных ресурсов.

Подход 1: инкрементальный MERGE из таблиц только для добавления

Этот подход использует инкрементальную стратегию merge для добавления и обновления строк (upsert: insert + update) в целевой таблице. Большинство платформ данных поддерживают стратегию merge. Подробнее см. в разделе поддерживаемых инкрементальных стратегий по адаптерам.

«Таблицы только для добавления» (append-only) — это подход к хранению данных, при котором в исходные данные непрерывно добавляются новые строки без обновлений и удалений.

Когда использовать стратегию merge

Используйте этот подход, когда необработанные события непрерывно поступают в промежуточную таблицу и вам нужна таблица фактов, обновляемая каждые несколько минут для работы почти в реальном времени.

Пример модели

В этом примере предположим, что необработанные события непрерывно поступают в raw.events (с помощью Snowpipe, Databricks Auto Loader, Kafka или аналогичного механизма загрузки) и вам нужна таблица фактов analytics.fct_events, обновляемая каждые несколько минут для работы почти в реальном времени.

Настройте SQL-модель следующим образом:

Используйте фильтр incremental, чтобы сканировать только строки с временной меткой новее самой поздней временной метки в целевой таблице.
Используйте incremental_strategy='merge' с unique_key=event_id для идемпотентного добавления и обновления строк (upsert).
Кластеризация по дате с помощью cluster_by=['event_date'] помогает исключать ненужные данные из сканирования во время операций MERGE (синтаксис зависит от хранилища).
Запускайте модель каждые несколько минут, чтобы обеспечить соглашение об уровне обслуживания (SLA) по актуальности данных, измеряемое в минутах, с учётом загрузки данных и расписания заданий.

В следующем примере используется синтаксис SQL Snowflake (приведение типов через ::, timestamp_ntz, настройка cluster_by). Обязательно адаптируйте синтаксис SQL и кластеризации для своего хранилища.

models/fct_events.sql

{{ config(
    materialized = 'incremental',
    incremental_strategy = 'merge',         -- по умолчанию в Snowflake
    unique_key = 'event_id',
    cluster_by = ['event_date']             -- повышает производительность MERGE (синтаксис зависит от хранилища)
) }}

with source_events as (

    select
        event_id,
        event_ts::timestamp_ntz       as event_ts,  -- синтаксис Snowflake для приведения типов
        to_date(event_ts)             as event_date,
        user_id,
        event_type,
        payload
    from {{ source('raw', 'events') }}

    {% if is_incremental() %} 
      -- Получать только новые/изменённые строки с момента последней успешной загрузки
      where event_ts >
            (select max(event_ts) from {{ this }})
    {% endif %}

),

deduped as (

    -- необязательно: если исходный поток данных может содержать дубликаты
    select *
    from (
        select
            *,
            row_number() over (
                partition by event_id
                order by event_ts desc
            ) as _rn
        from source_events
    )
    where _rn = 1

)

select
    event_id,
    event_ts,
    event_date,
    user_id,
    event_type,
    payload
from deduped;
Сообщить об ошибке в коде

Для достижения наилучших результатов:

Продуманно выбирайте ключи кластеризации для повышения производительности MERGE.
Отслеживайте производительность MERGE по мере роста таблицы.
Рассмотрите возможность добавления окна ретроспективной обработки (например, event_ts > max(event_ts) - interval '1 hour') для обработки данных, поступающих с задержкой.

Подход 2: CDC с использованием Snowflake Streams

Этот подход использует встроенные возможности отслеживания изменений данных (Change Data Capture, CDC) в Snowflake с помощью Streams — специфичной для Snowflake функции, которая отслеживает изменения (вставки, обновления, удаления) в исходных таблицах.

Когда использовать CDC

Используйте CDC, если:

В исходных таблицах часто обновляются данные (а не только добавляются новые строки).
Необходимо отслеживать как новые записи, так и изменения существующих записей.
Вы хотите избежать полного сканирования больших исходных таблиц.

Этот подход сочетает CDC на уровне хранилища и инкрементальную модель. Он не использует снимки. Рекомендации по выбору между инкрементальными моделями и снимками (а также по их совместному использованию) см. в разделе Выбор между инкрементальными моделями и снимками.

Настройка

Чтобы использовать этот подход, настройте поток в своём хранилище данных, а затем создайте модель для чтения этого потока.

Создайте поток (однократно, вне dbt):

create or replace stream RAW.EVENTS_STREAM
on table RAW.EVENTS;
Сообщить об ошибке в коде

Создайте модель, читающую поток:

models/fct_events_cdc.sql

{{ config(
    materialized = 'incremental',
    incremental_strategy = 'merge',
    unique_key = 'event_id',
    cluster_by = ['event_date'],
    snowflake_warehouse = 'TRANSFORM_WH'
) }}

with changes as (

    select
        event_id,
        event_ts::timestamp_ntz        as event_ts,
        to_date(event_ts)              as event_date,
        user_id,
        event_type,
        payload,
        metadata$action                as change_type
    -- ссылается на STREAM, а не на таблицу
    from {{ source('raw', 'events_stream') }}

),

filtered as (

    select *
    from changes
    where change_type in ('INSERT', 'UPDATE')
    -- Если нужно физически удалять записи, здесь также можно обработать 'DELETE'

)

select
    event_id,
    event_ts,
    event_date,
    user_id,
    event_type,
    payload
from filtered;
Сообщить об ошибке в коде

Особенности подхода

Есть несколько ключевых отличий от подхода 1:

Потоки возвращают только изменённые строки, поэтому фильтр по времени с is_incremental() не нужен. Каждый запуск обрабатывает только изменения, доступные на этот момент.
Запускайте модель каждые несколько минут, чтобы получать новые изменения и применять их к fct_events с помощью merge.
В результате вы получаете конвейер на основе CDC. Snowflake Streams отслеживает изменения, а dbt выполняет преобразования, тесты и отслеживание происхождения данных.

Подход 3: микропакетная обработка больших таблиц временных рядов

Для больших таблиц фактов (fact), в которых дозагрузка исторических данных или обработка за длительные ретроспективные периоды вызывает сложности, используйте incremental_strategy='microbatch' (доступно в dbt версии 1.9 и выше, а также в канале релизов v1 Latest в dbt platform). Подробнее см. в разделе инкрементальной микропакетной обработки. Обратите внимание, что Microsoft Fabric пока не поддерживает микропакетную обработку. Подробнее см. в разделе инкрементальных стратегий по адаптерам.

Для microbatch необходим event_time

Для каждой вышестоящей модели, передающей данные в эту микропакетную модель, также необходимо настроить event_time, чтобы dbt мог применять фильтры по времени к вышестоящим моделям. Иначе каждый пакет может приводить к повторному полному сканированию вышестоящих таблиц.

Когда использовать микропакетную обработку

У вас очень большие таблицы временных рядов (миллиарды строк).
Дозагрузка исторических данных при традиционных инкрементальных подходах выполняется медленно и сопряжена с рисками.
Вам нужно повторно обрабатывать данные частями удобного размера.
Данные часто поступают с задержкой.

Настройка модели

Предположим, у вас есть таблица fact_events со столбцом event_ts, и вы хотите обрабатывать её часовыми пакетами. Модель можно настроить следующим образом:

models/fct_events_microbatch.sql

{{ config(
    materialized        = 'incremental',
    incremental_strategy= 'microbatch',
    event_time          = 'event_ts',   -- столбец времени в этой модели
    batch_size          = 'hour',       -- обработка часовыми пакетами
    lookback            = 1,            -- повторная обработка 1 предыдущего пакета для учёта запоздавших данных
    unique_key          = 'event_id',
    cluster_by          = ['event_date'],
    full_refresh        = false
) }}

select
    event_id,
    event_ts::timestamp_ntz   as event_ts,
    to_date(event_ts)         as event_date,
    user_id,
    event_type,
    payload
from {{ ref('stg_events') }};
Сообщить об ошибке в коде

Основные особенности работы

Используйте микропакетную обработку для очень больших таблиц фактов (потоки кликов, IoT, данные кассовых систем) с многолетней историей.
Блок is_incremental() не нужен — dbt автоматически создаёт соответствующие предикаты WHERE event_ts BETWEEN.. для каждого пакета на основе event_time, batch_size, begin, lookback и других параметров.
Каждый запуск выполняет несколько небольших запросов (по одному на пакет), что делает масштабную дозагрузку исторических данных безопаснее и упрощает повторные попытки.
Параметр lookback автоматически обеспечивает обработку данных, поступающих с задержкой, за счёт повторной обработки недавних пакетов.
Планируйте задания с учётом своего SLA.

Выбор подходящего инкрементального подхода

Выбор подхода зависит от вашей задачи. Начните с подхода 1 (MERGE), поскольку он подходит для большинства случаев. Переходите к подходу 2 (используйте встроенные возможности CDC вашего хранилища данных), когда требуется эффективное отслеживание изменений. Выбирайте подход 3 (микропакетная обработка) при работе с очень большими объёмами данных.

Используйте следующую таблицу, чтобы выбрать подходящий подход:

Подход	Лучше всего подходит для	Основное преимущество
`merge` из таблиц только для добавления	Большинство стандартных задач	Простота и широкая известность
CDC с использованием Streams	Таблицы с частыми обновлениями	Эффективное отслеживание изменений
Микропакетная обработка	Очень большие таблицы временных рядов	Безопасная дозагрузка исторических данных и обработка запоздавших данных

Связанная документация

Нашли ошибку?

Создать GitHub Issue

Инкрементальные подходы для обработки данных почти в реальном времени

Подход 1: инкрементальный MERGE из таблиц только для добавления

Когда использовать стратегию merge

Пример модели

Подход 2: CDC с использованием Snowflake Streams

Когда использовать CDC

Настройка

Особенности подхода

Подход 3: микропакетная обработка больших таблиц временных рядов

Когда использовать микропакетную обработку

Настройка модели

Основные особенности работы

Выбор подходящего инкрементального подхода

Связанная документация

Нашли ошибку?

Начните работать с dbt.

Ресурсы dbt

Продукты

Сообщество

Поддержка

Наши соцсети

Подход 1: инкрементальный MERGE из таблиц только для добавления​

Когда использовать стратегию merge​

Пример модели​

Подход 2: CDC с использованием Snowflake Streams​

Когда использовать CDC​

Настройка​

Особенности подхода​

Подход 3: микропакетная обработка больших таблиц временных рядов​

Когда использовать микропакетную обработку​

Настройка модели​

Основные особенности работы​

Выбор подходящего инкрементального подхода​

Связанная документация​

Нашли ошибку?

Ресурсы dbt

Продукты

Сообщество

Поддержка

Наши соцсети

Подход 1: инкрементальный MERGE из таблиц только для добавления

Когда использовать стратегию merge

Пример модели

Подход 2: CDC с использованием Snowflake Streams

Когда использовать CDC

Настройка

Особенности подхода

Подход 3: микропакетная обработка больших таблиц временных рядов

Когда использовать микропакетную обработку

Настройка модели

Основные особенности работы

Выбор подходящего инкрементального подхода

Связанная документация