Data Engineering Design Patterns, глава 2: Паттерны загрузки данных

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

Инженер редко создаёт данные сам. Он берёт их у провайдеров: из других команд, других компаний, из API. Если загрузка сделана плохо, потребители получат неполные или сломанные данные. Потом придётся делать backfill.

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

Паттерн	Когда применять
Full Loader	Набор небольшой или нет признака изменений. Нужна полная копия каждый раз
Incremental Loader	Набор растёт. Есть колонка-дельта или партиции по времени
Change Data Capture (CDC)	Нужна задержка в секундах и учёт физических удалений
Passthrough Replicator	Нужна точная копия данных между средами
Transformation Replicator	Копия нужна, но часть полей (PII) нельзя переносить
Compactor	Слишком много маленьких файлов, листинг файлов тормозит чтение
Readiness Marker	Нужно знать, что набор данных полностью готов
External Trigger	Данные приходят нерегулярно. Нужен запуск по событию

Full Load: полная загрузка

Full Loader

Проблема. Для Silver-слоя нужен справочник устройств от внешнего провайдера. Он меняется несколько раз в неделю, в нём до миллиона строк. В данных нет колонки, по которой видно изменённые строки.

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

Простейший вариант - два шага: extract и load (EL). Это passthrough job. Данные идут из источника в приёмник без изменений.
Для однотипных хранилищ хватает встроенных команд экспорта и импорта или копирования файлов.
Для разных хранилищ нужна тонкая трансформация между шагами. Тогда это ETL, и обычно используют фреймворк обработки (Spark).

Последствия:

Объём данных. Если набор быстро растёт, статичные ресурсы не справятся. Помогает автомасштабирование.
Согласованность. Подход drop-and-insert опасен. Потребитель может прочитать частичные данные или вообще ничего.
Откат. После полной перезаписи нельзя вернуть старую версию, если нет time travel.

Как снизить риск.

Использовать транзакции.
Или использовать формат с версиями: Delta Lake, Iceberg.
Или скрыть данные за единым слоем доступа (view). Загружать в таблицу с версией, потом переключать view.

-- Загрузка в версионную таблицу
COPY devices_20240101 FROM '/data/devices.csv' CSV HEADER;

-- Переключение: потребители видят либо старую, либо новую версию
CREATE OR REPLACE VIEW devices AS SELECT * FROM devices_20240101;

Другие варианты реализации:

aws s3 sync с ключом --delete для переноса между бакетами.
Spark: чтение JSON и запись в таблицу Delta Lake.
Airflow: задача загрузки и задача переключения view.

Incremental Load: инкрементальная загрузка

Incremental Loader

Проблема. Часть событий посещений всё ещё пишется в транзакционную базу устаревшими продюсерами. Их нужно переносить в Bronze. Данных всё больше, поэтому каждый раз брать всё нельзя. События неизменяемые.

Как работает. Есть два варианта:

Колонка-дельта. Загрузчик помнит последнее значение (например, ingestion_time) и берёт только новые строки.
Партиции по времени. Загрузчик берёт целую партицию, например за прошлый час. Состояние хранить не нужно. Перед чтением нужен Readiness Marker.

# Колонка-дельта с границами окна: повторный запуск даёт тот же результат
batch = df.filter(
    f"ingestion_time BETWEEN '{date_from}' AND '{date_to}'"
)
batch.write.mode("append").save(output_path)

Последствия:

Physical deletes. Удалённая строка исчезает, и дельта о ней ничего не знает. Решения: soft delete (продюсер ставит флаг через UPDATE) или insert-only таблицы. Во втором случае потребитель сам определяет удаления и изменения.
Backfill. При перезапуске за два месяца загрузка станет полной и потребует больше ресурсов. Ограничьте окно, например один час. Тогда ресурсы предсказуемы, а backfill можно делать параллельно. В партиционной версии проблемы нет.
Риск event time. Если колонка-дельта - время события, поздние данные будут пропущены.

Примеры:

Оркестратор Airflow: FileSensor ждёт партицию, потом запускается Spark-задача.
Для колонки-дельты сенсор не нужен. Задача сразу фильтрует нужное окно.
Аргументы партиции берут из неизменяемого времени запуска ({{ ds }}). Это упрощает backfill.

Change Data Capture

Проблема. Загрузчик с расписанием слишком медленный. Потребители жалуются на ожидание. Нужно передавать каждое изменение таблицы в брокер за 30 секунд.

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

Читается commit log базы. Это append-only журнал всех операций.
Изменения идут в стриминг-брокер или другой приёмник.
Потребитель решает, хранить ли всю историю или только последнюю версию строки.
Перехватываются все операции, включая физические удаления. Просить soft delete у продюсера не нужно.

Последствия:

Риск	Что значит
Сложность	Нужна настройка базы и часто помощь ops: включить журнал, создать пользователя
Область данных	Клиент часто видит изменения только после своего запуска. Для истории комбинируйте с другими паттернами
Payload	Вместе со строкой приходят метаданные: тип операции, время, типы колонок. Лишнее нужно игнорировать
Семантика	Загруженные статичные строки превращаются в данные в движении. `JOIN` двух потоков может не найти пару: она ещё не пришла

Примеры:

Debezium с Kafka Connect. Настройка идёт через конфиг. Для PostgreSQL нужен logical replication с плагином pgoutput. Каждая таблица попадает в свой топик.
Delta Lake: встроенный change data feed. Включается свойством enableChangeDataFeed. В выдаче есть служебные колонки _change_type, _commit_version, _commit_timestamp. Обновления дают пары update_preimage и update_postimage.

Replication: репликация

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

Passthrough Replicator

Проблема. Есть среды dev, staging и prod. Справочник загружается в prod из внешнего API. API не идемпотентен: на один вызов в течение дня возвращает разные ответы. Поэтому в dev и staging нельзя просто запустить ту же загрузку. Нужны те же данные, что в prod.

Как работает. Два способа:

Уровень вычислений: задача EL, которая копирует файлы или строки без изменений.
Уровень инфраструктуры: политика репликации у облачного провайдера. Он копирует записи сам (например, replication rule для S3, MirrorMaker для Kafka).

Последствия:

Простота. Используйте самую простую команду копирования. Для JSON читайте как сырой текст, а не парсите. Если важны имена или число файлов, не берите распределённый фреймворк без нужных настроек.
Безопасность. Используйте push: среда-владелец сама отправляет данные в другие. Тогда ошибка не дестабилизирует целевую среду.
PII. Если в данных есть персональные поля, нужен Transformation Replicator.
Задержка. У инфраструктурной репликации часто есть SLA. Проверьте его.
Метаданные. Одних Parquet-файлов Delta Lake недостаточно, нужен и журнал. В Kafka важны ключи, заголовки и порядок внутри партиции.

Transformation Replicator

Проблема. Перед релизом задачи нужно тестировать на реальных данных. Синтетика не воспроизводит проблемы качества провайдера. Но в данных есть PII, и вне prod его нельзя использовать.

Как работает. Между чтением и записью добавляется трансформация. Она убирает или заменяет чувствительные поля. Варианты:

Функция в Spark или Flink.
SQL SELECT.
Замена значений (например, паттерном Anonymizer).

-- Уменьшение набора: убрать поля с PII
SELECT * EXCEPT (ip, latitude, longitude) FROM visits;

-- Или выдать только разрешённые колонки
GRANT SELECT (visit_id, event_time, user_id) ON TABLE visits TO user_a;

Последствия:

Риск сломать данные выше, чем в Passthrough. Пример: схема задаёт формат времени не такой, как в данных. Задача на staging падает. Совет: читайте спорные колонки как строки.
Десинхронизация. Сегодняшний список PII может устареть. Появятся новые поля, или старые станут PII. Используйте каталог данных или data contract с тегами чувствительных полей. Тогда правила применяются автоматически.

Data Compaction: уплотнение данных

Compactor

Проблема. Стриминговый пайплайн пишет события в object store. Через три месяца batch-задачи тратят 70% времени на листинг файлов и 30% на обработку. Маленьких файлов слишком много.

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

Много маленьких файлов объединяются в большие. Операций ввода-вывода становится меньше.
В Delta Lake это команда OPTIMIZE. В Iceberg это действие rewrite data files.
В Hudi с таблицей merge-on-read изменения хранятся строками. Compaction сливает их с колоночным хранилищем.
В Kafka это log compaction. Остаётся последняя запись по каждому ключу. Это не гарантирует уникальность ключей.

Последствия:

Цена и производительность. Compaction - обычная задача. На больших таблицах она дорогая. Запускайте её редко, лучше вне рабочих часов. Но тогда потребители какое-то время читают неуплотнённые данные. Иногда compaction стоит включить прямо в загрузку.
Согласованность. Compaction переписывает данные, которые уже читают. Для форматов с ACID (Delta Lake, Iceberg) это безопасно. Для сырых JSON и CSV сложнее.
Очистка. Старые файлы остаются. Нужна отдельная очистка (VACUUM). После неё нельзя откатиться к версии на удалённых файлах. Не ставьте слишком короткий срок хранения, иначе можно удалить файлы, которые ещё пишутся.

table = DeltaTable.forPath(spark, table_dir)
table.optimize().executeCompaction()   # уплотнить
table.vacuum()                         # удалить старые файлы

Data Readiness: готовность данных

Readiness Marker

Проблема. Каждый час готовится Silver-набор: исправлены проблемы качества, добавлен контекст из API. На него опираются ML и BI. Команды жалуются на неполные данные и просят сигнал готовности. Прямо запустить их пайплайны вы не можете: они в других командах.

Как работает. Набор помечается как готовый. Есть два варианта:

Flag file. После успешной записи создаётся файл, например _SUCCESS. Spark создаёт его сам для сырых форматов. В Delta Lake роль маркера играет commit log. Иначе файл создаёт оркестратор отдельной последней задачей.
Партиции. Если данные пишутся по часам, партиция 10 готова, когда началась партиция 11.

# Airflow: ждём маркер, не занимая слот воркера
FileSensor(filepath=f"{path}/_SUCCESS", mode="reschedule")

Последствия:

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

Event Driven: загрузка по событию

External Trigger

Проблема. Бэкенд выпускает новые функции не чаще раза в неделю. Справочник функций обновляется ежедневно, даже когда изменений нет. Это тратит ресурсы. Бэкенд при релизе шлёт событие в шину. Нужно запускать пайплайн только при новом релизе.

Как работает. Три действия:

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

Если запускать нечего, загрузку можно выполнить прямо в обработчике. Это event-driven ingestion.

Последствия:

Push или pull. Pull-обработчик работает постоянно и часто опрашивает пустой источник. Push-обработчик спит, пока продюсер не пришлёт сообщение. Push экономнее.
Контекст запуска. Обработчик не должен быть просто пингом. Передайте метаданные: кто запустил, конверт уведомления, время обработки, время события. Они помогут при разборе сбоев.
Ошибки. Событие нельзя потерять. Проектируйте обработчик под сбои. Используйте паттерны главы 3, например Dead-Letter.

Пример. Облачные serverless-функции (AWS Lambda и другие) подходят как триггер. Функция запускается при появлении объекта в бакете и вызывает API Airflow. У DAG стоит schedule_interval=None, и он стартует только от явного вызова.

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

Термин	Значение
Passthrough job	Задача `EL`: данные проходят без изменений
Delta column	Колонка для поиска новых или изменённых строк
Soft delete	Удаление через флаг, а не физически
Insert-only table	Таблица, куда только добавляют строки
Commit log	Журнал всех изменений базы
CDC	Захват изменений из журнала базы
Change data feed	Встроенный CDC в Delta Lake
PII	Персональные данные, которые нельзя свободно копировать
Small files problem	Много маленьких файлов замедляют листинг и чтение
Compaction	Объединение маленьких файлов в большие
`VACUUM`	Очистка файлов, которые больше не нужны
Flag file	Файл-маркер готовности, например `_SUCCESS`
Push / pull	Продюсер уведомляет сам / потребитель сам проверяет

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

Начинайте с Full Loader для небольших справочников. Это самое простое решение.
Для больших и растущих наборов используйте партиции по времени. Состояние хранить не нужно, backfill проще.
Ограничивайте окно загрузки верхней границей. Тогда повторный запуск даёт тот же результат.
Не делайте drop-and-insert в таблице, которую читают. Используйте транзакции или переключение view.
CDC включайте, когда нужны секунды и удаления. Заранее договоритесь с командой базы о журнале.
Копируйте prod-данные в dev только через Transformation Replicator с учётом PII.
Запланируйте compaction и очистку для таблиц от стриминга.
Не запускайте зависимые пайплайны по времени. Используйте маркер готовности или событие.

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

Когда Incremental Loader на колонке-дельте не подходит и что взять вместо него?
Почему drop-and-insert опасен и как этого избежать?
Чем CDC отличается от Incremental Loader по задержке, удалениям и сложности?
Когда нужен Transformation Replicator, а не Passthrough Replicator?

Предыдущая: Введение в паттерны. Следующая: Паттерны обработки ошибок.