Лямбда-представления для дашбордов с данными почти в реальном времени

Далее — примеры для Snowflake

На этой странице в примерах кода используется Snowflake, но вы можете адаптировать паттерн лямбда-представлений для других хранилищ данных.

Паттерн лямбда-представлений объединяет таблицу фактов, обновляемую пакетно или инкрементально, с небольшим срезом самых свежих данных, поступающих почти в реальном времени (near real-time, NRT), и предоставляет доступ к ним через единое представление. Это устаревший, но всё ещё полезный паттерн, который некоторые команды использовали для создания оперативных дашбордов с данными почти в реальном времени на базе dbt и хранилища данных.

Когда использовать лямбда-представления

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

Исходные условия

Примеры на этой странице предполагают следующую конфигурацию:

• Необработанные события непрерывно поступают в raw.events с помощью механизма потоковой загрузки вашего хранилища данных (например, Snowpipe, Databricks Auto Loader, Kafka или аналогичного механизма загрузки).
• У вас уже есть инкрементальная таблица фактов, которая обновляется каждые несколько минут с использованием incremental_strategy='merge'.
• Большинству дашбордов достаточно данных из этой инкрементальной таблицы, но небольшой группе оперативных дашбордов нужны данные «на текущий момент» (например, события за последние несколько минут).

Как работает этот паттерн

• Базовая инкрементальная таблица обновляется каждые несколько минут с использованием incremental_strategy='merge'.
• NRT-представление выбирает только события новее максимального значения event_ts, уже сохранённого в базовой инкрементальной таблице.
• Лямбда-представление объединяет базовую таблицу и NRT-представление с помощью UNION ALL, устраняя дубликаты строк на основе семантики первичного ключа.

Нижележащие BI-инструменты и дашборды обращаются только к лямбда-представлению.

Базовая инкрементальная таблица

В качестве базовой таблицы можно использовать инкрементальную модель с merge из паттерна 1 для Snowflake; для полноты приведём её здесь:

-- models/fct_events.sql
{{ config(
    materialized = 'incremental',
    incremental_strategy = 'merge',
    unique_key = 'event_id',
    cluster_by = ['event_date'],
    snowflake_warehouse = 'TRANSFORM_WH'
) }}

with source_events as (

    select
        event_id,
        event_ts::timestamp_ntz       as event_ts,
        to_date(event_ts)            as event_date,
        user_id,
        event_type,
        payload
    from {{ source('raw', 'events') }}

    {% if is_incremental() %}
      -- Only pull new/changed rows since last successful load
      where event_ts >
            (select max(event_ts) from {{ this }})
    {% endif %}

)

select *
from source_events;

Сообщить об ошибке в коде

Настройте запуск этой модели, например, каждые 5–15 минут в составе задания для обработки данных почти в реальном времени.

NRT-представление: строки новее данных в базовой таблице

NRT-представление возвращает только события со значением event_ts, превышающим максимальную временную метку в базовой таблице, поэтому пересечения данных и двойного учёта не возникает:

-- models/fct_events_nrt.sql
{{ config(
    materialized = 'view'
) }}

with base_max as (
    select max(event_ts) as max_event_ts
    from {{ ref('fct_events') }}
),

fresh_events as (
    select
        e.event_id,
        e.event_ts::timestamp_ntz   as event_ts,
        to_date(e.event_ts)         as event_date,
        e.user_id,
        e.event_type,
        e.payload
    from {{ source('raw', 'events') }} as e
    cross join base_max
    where e.event_ts > base_max.max_event_ts
)

select *
from fresh_events;

Сообщить об ошибке в коде

Особенности:

• Расписание не требуется — это обычное представление над raw.events с фильтрацией по max(event_ts) из fct_events.
• Каждый запрос к fct_events_nrt сканирует только данные, поступившие «с момента последнего пакетного обновления»: это должен быть небольшой временной интервал (например, несколько минут или часов в зависимости от частоты запуска задания).

Лямбда-представление: единая точка чтения данных для BI

Лямбда-представление объединяет исторические данные из базовой инкрементальной таблицы с самыми свежими событиями из NRT-представления.

-- models/fct_events_lambda.sql
{{ config(
    materialized = 'view'
) }}

select
    event_id,
    event_ts,
    event_date,
    user_id,
    event_type,
    payload
from {{ ref('fct_events') }}

union all

select
    event_id,
    event_ts,
    event_date,
    user_id,
    event_type,
    payload
from {{ ref('fct_events_nrt') }};

Сообщить об ошибке в коде

Настройте BI-инструменты и дашборды на чтение из analytics.fct_events_lambda. Большая часть данных поступает из предварительно рассчитанной инкрементальной таблицы, а самые свежие события (с момента последнего запуска dbt) — из запроса к raw.events, выполняемого в момент обращения.

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

Что следует учитывать

При использовании этого паттерна учитывайте следующее:

• Структура затрат
  • Каждый запрос к fct_events_lambda должен читать NRT-срез из raw.events в дополнение к базовой таблице.
  • Используйте этот паттерн только для действительно оперативных дашбордов, потребности которых оправдывают дополнительные затраты на каждый запрос.
• Свежесть данных
  • Свежесть данных ограничена:
    • Частотой запуска инкрементального задания dbt (давностью обновления fct_events), а также
    • Задержкой загрузки в raw.events (Snowpipe / слой потоковой загрузки).
• Сложность в сравнении с альтернативами
  • Во многих современных реализациях на Snowflake динамическая таблица или материализованное представление с небольшим значением target_lag могут обеспечить аналогичные соглашения об уровне обслуживания по принципу «данные всегда не старше X минут», при этом требуя меньше собственного SQL-кода и передавая управление инкрементальной логикой хранилищу данных.
  • Лямбда-представления лучше рассматривать как продвинутый / устаревший паттерн, который по-прежнему можно использовать, если вы:
    • Хотите сохранить всю логику в SQL-коде dbt
    • Не располагаете нужной возможностью хранилища данных в своей среде
    • Расширяете существующую реализацию, уже построенную таким образом

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

Создать GitHub Issue