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

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

1. [Инкрементальный MERGE из таблиц только для добавления](#incremental-merge-from-append-only-tables)
2. [CDC с использованием Snowflake Streams](#cdc-with-snowflake-streams)
3. [Микропакетная обработка больших таблиц временных рядов](#microbatch-for-large-time-series-tables)

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

Некоторые подходы в этом руководстве используют возможности, специфичные для Snowflake. В других хранилищах есть аналогичные возможности с другими реализациями. Документацию для конкретных адаптеров см. в разделе [дополнительных ресурсов](https://docs.getdbt.tech/best-practices/how-we-handle-real-time-data/3-warehouse-native-features.md#resources-by-warehouse).

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

Этот подход использует инкрементальную стратегию `merge` для добавления и обновления строк (upsert: insert + update) в целевой таблице. Большинство платформ данных поддерживают стратегию `merge`. Подробнее см. в разделе [поддерживаемых инкрементальных стратегий по адаптерам](https://docs.getdbt.tech/docs/build/incremental-strategy.md#supported-incremental-strategies-by-adapter).

«Таблицы только для добавления» (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

```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](https://docs.snowflake.com/en/user-guide/streams-intro) — специфичной для Snowflake функции, которая отслеживает изменения (вставки, обновления, удаления) в исходных таблицах.

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

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

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

Этот подход сочетает CDC на уровне хранилища и инкрементальную модель. Он не использует снимки. Рекомендации по выбору между инкрементальными моделями и снимками (а также по их совместному использованию) см. в разделе [Выбор между инкрементальными моделями и снимками](https://docs.getdbt.tech/best-practices/how-we-handle-cdc/2-choosing-incremental-or-snapshots.md).

### Настройка

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

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

```sql
create or replace stream RAW.EVENTS_STREAM
on table RAW.EVENTS;
```

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

models/fct\_events\_cdc.sql

```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](#incremental-merge-from-append-only-tables):

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

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

Для больших таблиц фактов (`fact`), в которых дозагрузка исторических данных или обработка за длительные ретроспективные периоды вызывает сложности, используйте `incremental_strategy='microbatch'` (доступно в dbt версии 1.9 и выше, а также в канале релизов **v1 Latest** в dbt platform). Подробнее см. в разделе [инкрементальной микропакетной обработки](https://docs.getdbt.tech/docs/build/incremental-microbatch.md). Обратите внимание, что Microsoft Fabric пока не поддерживает микропакетную обработку. Подробнее см. в разделе [инкрементальных стратегий по адаптерам](https://docs.getdbt.tech/docs/build/incremental-strategy.md#supported-incremental-strategies-by-adapter).

Для microbatch необходим event\_time

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

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

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

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

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

models/fct\_events\_microbatch.sql

```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](#incremental-merge-from-append-only-tables) (`MERGE`), поскольку он подходит для большинства случаев. Переходите к [подходу 2](#cdc-with-snowflake-streams) (используйте встроенные возможности CDC вашего хранилища данных), когда требуется эффективное отслеживание изменений. Выбирайте [подход 3](#microbatch-for-large-time-series-tables) (микропакетная обработка) при работе с очень большими объёмами данных.

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

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

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

* [Инкрементальные модели](https://docs.getdbt.tech/docs/build/incremental-models-overview.md)
* [Микропакетные инкрементальные модели](https://docs.getdbt.tech/docs/build/incremental-microbatch.md)
* [Настройка инкрементальных моделей в dbt](https://docs.getdbt.tech/docs/build/incremental-models.md)