Data Engineering Design Patterns, глава 5: Паттерны добавления ценности данным

Главная мысль: Сырые данные сами по себе почти бесполезны. Ценность появляется, когда их дополняют, сжимают в понятный вид или доставляют в нужном порядке.

Главы 3 и 4 были про надёжность. Но надёжный пайплайн ещё не создаёт полезный датасет. Эта глава показывает, как сделать данные полезнее. Есть два пути. Первый: добавить информацию (enrichment, decoration). Второй: убрать лишнее и получить обзор (aggregation, sessionization). Отдельная тема: порядок записей (ordering).

Все паттерны главы

Паттерн	Когда применять
`Static Joiner`	Нужно добавить к событиям медленно меняющийся справочник
`Dynamic Joiner`	Обе стороны join постоянно меняются (два потока)
`Wrapper`	Нужно сохранить исходную запись и рядом положить вычисленные поля
`Metadata Decorator`	Нужно добавить технический контекст, но скрыть его от бизнес-пользователей
`Distributed Aggregator`	Связанные записи лежат на разных машинах или в разных хранилищах
`Local Aggregator`	Данные уже разложены по ключу, и shuffle не нужен
`Incremental Sessionizer`	Сессии строятся пакетно, по партициям
`Stateful Sessionizer`	Сессии нужны почти в реальном времени
`Bin Pack Orderer`	Нужен порядок записей и bulk-запись при частичных коммитах
`FIFO Orderer`	Нужен порядок, и каждое событие нужно отправить сразу, без буфера

Data Enrichment: добавить контекст из другого датасета

Событие знает, кто и когда сделал действие. Но оно не знает профиль пользователя или его историю. Эти данные берут из другого источника.

Static Joiner

Проблема. Нужно связать активность пользователя с его профилем. Профиль лежит в статичном справочнике.

Как работает:

Выберите ключ для join, например user_id.
Соедините события со справочником обычным JOIN. Работает и в batch, и в streaming.
Если справочник меняется со временем, добавьте условие по времени. Справочник тогда ведётся как SCD type 2 (строки с датами действия) или type 4 (текущая таблица и отдельная таблица истории).
Данные из API берите пачками (bulk), а не по одной записи. Сеть здесь самая дорогая часть.

SELECT v.visit_id, v.page, u.email
FROM visits v
JOIN users u
  ON u.id = v.user_id
 AND v.event_time BETWEEN u.start_date AND u.end_date;

Компромиссы:

Справочник и события обновляются с разной скоростью. Join может не найти запись.
Для idempotency при backfill нужна история справочника. Если источник не умеет запросы по времени, загрузите его копию к себе и ведите как SCD.
Данные из API лучше сначала материализовать в таблицу. Тогда повторный запуск даёт тот же результат.
В streaming справочник и поток живут отдельно. Задание не ждёт обновления справочника. Используйте формат таблиц с атомарной записью, чтобы не прочитать пустой справочник.
Для batch можно дождаться справочника через Readiness Marker (глава 2).

Dynamic Joiner

Проблема. Профили меняются тысячи раз в неделю. Статичный справочник устарел. Изменения идут в брокер, например через CDC.

Как работает:

Обе стороны join являются потоками. У них разная задержка.
Быстрый поток ждёт медленный в буфере с ограничением по времени.
Старые записи удаляет GC watermark. Он очищает буфер от записей, которые уже никто не дождётся.
Фреймворки (Spark Structured Streaming, Flink) делают буфер сами. Вам нужно задать условие по времени и watermark.

visits.withWatermark("event_time", "10 minutes").join(
    ads.withWatermark("display_time", "10 minutes"),
    F.expr("""page = visit_page AND
              display_time BETWEEN event_time AND event_time + INTERVAL 2 minutes"""),
    "left_outer")

Компромиссы:

Размер буфера против полноты join. Больше буфер, больше совпадений, но выше стоимость.
Очень поздние записи после watermark теряются. Для них нужна работа с late data (глава 3).
Стопроцентной полноты join нет ни в одном из двух паттернов.
В Flink есть ещё temporal table join. Он берёт для события самую свежую версию справочника на момент события.

Data Decoration: подготовить запись к использованию

Wrapper

Проблема. Данные приходят от разных поставщиков в разных схемах. Нужно вычислить общие поля. Исходная запись должна остаться для отладки.

Как работает:

Запись «оборачивают» в конверт. Внутри исходные значения и вычисленные.
Источник вычисленных полей: сами данные или контекст запуска (версия job, время обработки).
Для таблиц есть 4 варианта: исходная строка как struct плюс вычисленные колонки; исходные колонки плюс вычисленный struct; все колонки на одном уровне; две таблицы, связанные ключом.
Первые два варианта денормализованы. Они быстрее читаются. Четвёртый нормализован. Он изолирует данные логически.

SELECT
  CASE WHEN context.user.connected_since IS NULL THEN false ELSE true END AS is_connected,
  CONCAT_WS('-', page, context.referral) AS page_referral_key,
  STRUCT(visit_id, event_time, user_id, page, context) AS raw
FROM input_visits;

Компромиссы:

Данные пользователя лежат в двух местах. Потребителям сложнее их искать.
Вычисленные поля увеличивают размер записи. В колоночных хранилищах помогает выбор нужных колонок.
Подходит для слоя Silver. Конечным пользователям лучше отдавать упрощённый вид.
Нужно следить за схемой (паттерны consistency в главе 9).

Metadata Decorator

Проблема. Нужно знать версию job, которая создала запись. Но пользователям эта информация не нужна. Лишние поля им мешают.

Как работает:

Техническую информацию кладут в слой метаданных хранилища.
В Kafka это headers записи. В объектных хранилищах это tags файла.
В реляционной БД метаданных нет. Тогда добавьте колонку processing_context и скройте её через view или права доступа. Или вынесите контекст в отдельную таблицу и свяжите по processing_context_id.
Бизнес-данные в метаданные не кладут.

Компромиссы:

Не все системы поддерживают метаданные. Например, у Kinesis Data Streams нет headers.
В таблицах нужны дополнительные колонки или таблицы, это больше работы.
Отличие от Wrapper: метаданные описывают данные и скрыты. Wrapper добавляет бизнес-атрибуты, и они видны.
Скрытые бизнес-поля никто не найдёт. Не прячьте их в метаданных.

Data Aggregation: уменьшить, чтобы увидеть картину

Distributed Aggregator

Проблема. Нужен OLAP-куб по событиям: количество, средняя длительность, срезы по географии и устройствам. Данные не помещаются на одну машину.

Как работает:

Данные группируют по ключу и применяют функцию reduce. Код такой же, как для небольшого набора.
Внутри фреймворк делает shuffle: пересылает записи по сети, чтобы все строки одного ключа оказались на одном узле.
Если агрегат допускает частичный расчёт (например count), каждый узел сначала считает локально. Потом пересылаются только числа.
Это модель MapReduce. Она развилась от Hadoop к Spark.
Паттерн работает и для данных в разных хранилищах: файлы плюс таблица в PostgreSQL, или external table в BigQuery.

Компромиссы:

shuffle часто является главным источником задержки. Проверяйте план запроса (explain(), узел Exchange).
Data skew: один ключ встречается намного чаще других. Один узел перегружен. Помогает salting: добавьте случайное число к ключу, агрегируйте дважды. В Spark помогает также Adaptive Query Execution.
Узел закончил свою часть, но его нельзя освободить: на нём лежат shuffle-данные для других. Решение: shuffle service. Он хранит и отдаёт эти данные сам, и узел можно освободить.

(df.withColumn("salt", (F.rand() * 3).cast("int"))
   .groupBy("group_key", "salt").agg(...)
   .groupBy("group_key").agg(...))

Local Aggregator

Проблема. Данные уже лежат в партициях по ключу. Число партиций не меняется. Хочется убрать дорогой shuffle.

Как работает:

Все записи одного ключа уже находятся в одной партиции. Задача агрегирует их на месте.
Задачи изолированы. Они не ждут друг друга.
Работа производителя: писать запись по статичному ключу и держать число партиций неизменным.
Работа потребителя: в Kafka Streams есть groupByKey. В Spark помогают mapPartitions, foreachPartition и одинаково разложенные buckets.
В AWS Redshift похожий эффект даёт DISTSTYLE KEY и DISTSTYLE ALL (маленькую таблицу копируют на все узлы).

Компромиссы:

Масштабирование сложное. Если изменить число партиций, группы развалятся. Нужна полная перекладка данных.
Нужен один ключ группировки для всех потребителей. Если кому-то нужна другая группировка, для него остаётся Distributed Aggregator.
Для небольших данных паттерн не нужен: набор помещается в память одного узла.

Sessionization: события в сессии

Сессия объединяет события одной активности: начало, действия, конец. Конец обычно определяется периодом неактивности.

Incremental Sessionizer

Проблема. События лежат в часовых партициях. Сессия заканчивается после 2 часов без активности. Одна сессия может растянуться на 3 партиции.

Как работает:

Нужны три хранилища: входные данные, завершённые сессии, незавершённые (pending) сессии.
Pending-хранилище закрытое. Формат в нём может отличаться от публичного.
Каждый запуск соединяет новые события с pending-сессиями прошлого запуска. Получается новая сессия, дополненная сессия или сессия без новых данных.
Логика состоит из трёх состояний: инициализация, накопление, завершение.
Завершённые сессии идут в публичный вывод. Остальные остаются в pending.
Idempotency: перед запуском удалите готовые и pending-сессии этого запуска и всех следующих (DELETE ... WHERE execution_time_id >= '{{ ds }}'). Потом сессии строятся заново.

Компромиссы:

Длинный период неактивности включает больше поздних данных, но требует больше ресурсов.
Частичные сессии можно отдавать раньше. Помечайте их, например is_completed = false.
Сессия зависит от прошлых партиций. Поздние данные или backfill одной партиции заставляют пересчитать все следующие.
Простая схема: пересчитать всё. Умная схема: искать только затронутые сущности. Она дешевле, но сложнее.

Stateful Sessionizer

Проблема. Сессии нужны быстрее часа. События в брокере доступны через секунды.

Как работает:

Тот же процесс, что в Incremental Sessionizer, но в streaming.
Роль pending-хранилища играет state store. Он живёт в памяти и копируется в надёжное хранилище.
Два способа: session window (окно для ключа с gap duration) или arbitrary stateful processing (свой код и свои правила истечения).
Есть поддержка в Spark Structured Streaming, Flink и Dataflow.

Компромиссы:

Состояние сохраняется периодически (checkpointing). После сбоя часть событий обрабатывается повторно: at-least-once. Не используйте в ключе сессии значения, которые меняются между запусками.
Масштабирование требует перераспределения состояния. Это дороже, чем в stateless-задаче.
Истечение сессии лучше считать по event time (или watermark), а не по processing time. Processing time ломается при задержках.

Data Ordering: доставка в правильном порядке

Порядок важен, когда потребитель читает события хронологически. Угроза: частичный коммит (partial commit). Bulk API записывает только часть пачки. При повторе порядок ломается.

Bin Pack Orderer

Проблема. Нужно записать события каждому партнёру по минутам и в порядке event time. Хранилище использует bulk-запись с частичными коммитами. По одной записи слать дорого.

Как работает:

Отсортируйте записи по ключу и времени.
Разложите их по delivery bins так, чтобы в одном bin был только один экземпляр каждого ключа.
Отправляйте bins по очереди. Следующий bin идёт только после полной записи текущего.
Повтор bin не нарушает порядок: в нём у ключа одна запись.
В Spark сортировка идёт локально через sortWithinPartitions.

Компромиссы:

Порядок гарантирован только внутри одного запуска. Если упал весь пайплайн, повтор может нарушить порядок.
Реализация сложнее обычной сортировки: нужна своя логика bins.

FIFO Orderer

Проблема. Нужно отправить каждое событие сразу, без буфера. Порядок должен сохраниться.

Как работает:

Отправляйте запись за записью. Перед следующей ждите подтверждения.
Вариант с bulk: ограничьте параллелизм одним запросом (max.in.flight.requests.per.connection = 1).
В Kafka можно включить idempotent producer. Он допускает до 5 параллельных запросов и сохраняет порядок.
В Kinesis используется SequenceNumberForOrdering. В Pub/Sub используется ordering_key (в пределах одного региона).

Компромиссы:

Много запросов в сеть, выше задержка. Помогает несколько потоков, но порядок гарантируется только внутри одной сущности.
FIFO не значит exactly-once. Сбой между send и ack вызовет повтор. Нужны паттерны идемпотентности (глава 4).

Ключевые термины

Термин	Значение
Enrichment	Добавление данных из другого датасета
Decoration	Добавление вычисленных или технических полей к записи
`SCD` type 2 / type 4	Хранение истории справочника: строки с датами или отдельная таблица
`GC watermark`	Граница, после которой запись удаляется из буфера join
`Shuffle`	Пересылка записей между узлами по ключу
`Data skew`	Перекос: один ключ намного больше остальных
`Salting`	Добавление случайного значения в ключ, чтобы разбить перекос
Session window	Окно по ключу, которое закрывается после паузы `gap duration`
`State store`	Хранилище состояния stateful-задания
Partial commit	Bulk-запись, где часть пачки записана, а часть нет
Delivery bin	Пачка записей с уникальными ключами для безопасной bulk-доставки

Что это значит на практике

Справочник для join нужно версионировать. Тогда backfill даёт тот же результат. Подойдёт SCD2 или snapshot в dbt.
Не вызывайте внешний API на каждую строку. Берите пачками или материализуйте API в таблицу.
Смотрите план запроса. Найдите Exchange и перекос. Часто один ключ тормозит всё задание.
Храните технический контекст отдельно. Версия job и время обработки полезны при расследованиях, но не нужны в витрине.
Сессии зависят от соседних партиций. Планируйте backfill заранее. Он может быть дорогим.
В streaming считайте всё по event time. Задайте watermark и проверьте, как он влияет на потерю поздних записей.
Если нужен порядок, проверьте API приёмника. Узнайте, есть ли у него частичные коммиты.

Вопросы для самопроверки

Чем Static Joiner отличается от Dynamic Joiner? Какую роль играет GC watermark?
Почему shuffle замедляет Distributed Aggregator? Как Local Aggregator его избегает и что теряет?
Чем Incremental Sessionizer отличается от Stateful Sessionizer? Что делать с частичными сессиями?
Как Bin Pack Orderer сохраняет порядок при частичных коммитах? Почему FIFO не даёт exactly-once?

Предыдущая: Паттерны идемпотентности. Следующая: Паттерны потока данных.