Data Engineering Design Patterns, глава 6: Паттерны потока данных

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

Паттерны работают на двух уровнях:

Оркестрация (orchestration layer): задачи в Airflow, AWS Step Functions, Azure Data Factory. Здесь же связи между командами.
Обработка данных (processing layer): код в Spark, SQL. Здесь паттерны делают бизнес-логику понятнее.

Паттерны главы

Паттерн	Когда применять
`Local Sequencer`	Большая задача падает и каждый раз стартует с нуля. Нужно разбить её на шаги
`Isolated Sequencer`	Два пайплайна принадлежат разным командам, но один даёт данные другому
`Aligned Fan-In`	Одна задача ждёт успеха всех родителей (например, 24 часа в сутки)
`Unaligned Fan-In`	Задача должна стартовать, даже если часть родителей упала
`Parallel Split`	Один результат нужен нескольким независимым задачам
`Exclusive Choice`	Из нескольких веток нужно выбрать только одну
`Single Runner`	Запуск зависит от результата предыдущего. Одновременно идёт только один
`Concurrent Runner`	Запуски независимы. Их можно гнать параллельно

Последовательность (Sequence)

Здесь вопрос такой: где провести границы между шагами и между пайплайнами.

Local Sequencer

Проблема. Старый job вырос до сотен строк. Трансформаций стало в три раза больше. При любой ошибке он стартует с начала. Отлаживать его долго.

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

Большой компонент делят на малые шаги. Они идут друг за другом.
Порядок следует за зависимостью данных: задача B стартует после A.
Пример: загрузка делится на проверку готовности (Readiness Marker) и саму загрузку (Full Loader).
Цепочку можно строить в оркестраторе или внутри кода обработки. Выбор делают по трём критериям:

Критерий	Что смотреть
Separation of concerns	Если трудно придумать имя задаче, в ней слишком много
Maintainability	При retry пересчитываются все шаги до упавшего. Если проверка готовности вызывает 10 платных API, это дорого
Implementation effort	Оркестратор уже умеет запускать SQL или API. В одной задаче это придётся писать самому

Цена. Главный риск - неверные границы. Слишком много задач дают накладные расходы планировщика. Мало задач - долгие перезапуски.

Как выбрать границу:

Шаги, которые должны перезапускаться отдельно, делите.
Дорогую промежуточную стадию отделяйте от финальной трансформации. Тогда при сбое финала не нужно считать её заново.
Операции, которые должны быть одной транзакцией, оставляйте вместе. Пример: запись данных и обновление версии в state table.

# Airflow: цепочка из трёх задач
wait_for_data >> load_to_table >> expose_table

Если у шагов нет зависимостей и нет сбоев, обычного cron может быть достаточно.

Isolated Sequencer

Проблема. Ваша команда чистит и обогащает данные. Команда визуализации забирает их и строит свои представления. Один пайплайн объединить нельзя, потому что команды разные.

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

Границу проводят по роли: провайдер и потребитель. Если вы и то и другое, смотрите на сложность. Читаемость страдает - делите. Иначе используйте Local Sequencer.
Затем выбирают способ запуска:

Стратегия	Как работает	Связность
Data-based	Провайдер пишет данные и marker. Потребитель ждёт marker (`Readiness Marker`)	Слабая. Потребитель свободен в выборе данных
Task-based	Провайдер сам запускает пайплайн потребителя	Сильная. Переименование у потребителя ломает провайдера

Task-based подходит, когда общего датасета нет. Например, потребитель только шлёт уведомление.
В Airflow data-based делают через FileSensor. Task-based делают через ExternalTaskMarker и ExternalTaskSensor.

Цена:

Scheduling. Лучше одно расписание у обоих. Иначе одной стороне нужно пропускать часть запусков.
Communication. Нужна договорённость между командами. Если обе стороны не знают входы и выходы друг друга, изменение ломает цепочку.
Явная связь по данным не видна в графе. Помогает data lineage (глава 10).

Fan-In: много веток в одну

Aligned Fan-In

Проблема. Ежедневная агрегация по посещениям блога. Данные лежат партициями по часам. Потребители хотят полный день. Считать всё одним job нельзя: он слишком большой.

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

Строят ветки по часам. Каждая считает частичный результат.
Общая задача стартует, когда все ветки успешны.
На уровне обработки ветки соединяют так:
- UNION - добавляет строки, те же колонки (вертикально).
- JOIN - добавляет колонки (горизонтально).
Плюс первый: ошибка видна раньше. Ветка за последний час падает, а не весь суточный job.
Плюс второй: чинить и перезапускать нужно только упавший час, а не все 24.

Цена:

Пик нагрузки: 24 задачи стартуют одновременно. Ограничьте параллелизм или считайте по часам инкрементально.
Scheduling skew: общая задача ждёт самую медленную ветку.
Scheduling overhead: слишком мелкие задачи нагружают планировщик.
Чем больше разбиение, тем длиннее и сложнее пайплайн.

for hour in [f"{h:02d}" for h in range(24)]:
    sensor = wait_for_hour(hour)
    loader = load_hour(hour)
    clear >> sensor >> loader >> build_daily

Осторожно с UNION по позиции. В SQL колонки склеиваются по порядку, не по имени. a, b, c и c, b, a смешаются неверно. В Spark для этого есть unionByName.

Unaligned Fan-In

Проблема. Aligned Fan-In работает. Но иногда один час не обработан, и потребители не получают ничего. Договорились отдавать частичный результат и дозаполнять потом.

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

Условие запуска ослабляют. Дочерняя задача стартует даже при сбое части родителей.
Если часть родителей упала, но это допустимо, считаем по частичным данным.
Если упали все, можно запустить fallback или обработчик ошибок.
В Airflow это trigger_rule, например ALL_DONE. В Step Functions есть Catch.

Цена:

Readability. Условие спрятано в коде. В графе видно две ветки, и неясно, какая сработает.
Partial data. Потребитель должен знать, что данные неполные. Способы:
- таблица полноты (12 из 24 часов = 50%);
- флаг в самих данных, например is_approximate;
- тег в metadata хранилища;
- уведомление потребителям.
Можно не публиковать частичный результат, а записать его во внутреннюю таблицу.

-- Пометить строку как приблизительную, если часов меньше 24
CASE WHEN hours_count = 24 THEN false ELSE true END AS is_approximate

Fan-Out: одна ветка в несколько

Parallel Split

Проблема. Вы мигрируете старый C# framework на Python. Новому решению пока не доверяют. Старый пайплайн работает. Нужно писать результат в два места.

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

Общая родительская задача, затем две и больше независимые дочерние.
На уровне обработки общий датасет считают один раз. В Spark это persist(), в SQL - временная таблица.
Ветки не должны мешать друг другу. Общие переменные должны быть только для чтения.
Если ветки нагружают разные ресурсы, дайте каждой свою инфраструктуру (CPU-bound и memory-bound job).
При повторных запусках ветка не должна дублировать данные. Помогает идемпотентность из главы 4. В Delta Lake для этого есть txnAppId и txnVersion.

Цена:

Blocked execution. Следующий запуск ждёт самую медленную ветку. Медленную ветку можно вынести в отдельный пайплайн (Isolated Sequencer).
Hardware. Одна инфраструктура не подходит для разных профилей нагрузки.

Exclusive Choice

Проблема. С 1 января новая версия job должна работать только для новых дат. При backfill старые даты идут через старую версию. Новый пайплайн создавать не хочется: история запусков потеряется.

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

Перед ветвлением ставят задачу-условие. Она выбирает один путь.
В Airflow это BranchPythonOperator. В Azure Data Factory - if condition. В Step Functions есть Choice.
На уровне обработки это обычный if/else.
Условие лучше строить на metadata (дата запуска, параметр, схема файла). Условие по данным требует чтения и дороже.

Цена:

Complexity factory. Каждая ветка усложняет граф. Проверка: объясните пайплайн коллеге. Если объяснение длинное, пайплайн сложен.
Hidden logic. Условия внутри job легко забыть. Лучше вынести выбор в оркестратор и запускать разные точки входа.
Heavy conditions. Условие по данным увеличивает время job. Читайте данные один раз и по возможности инкрементально.
Для периода действия логики часто проще завести отдельный пайплайн с start_date и end_date.

Условие можно передать параметром job. Тогда выбор прячут в фабрике (Factory из software engineering), которая возвращает нужные SparkSession и writer.

Оркестрация (Orchestration)

Остался вопрос: сколько запусков одного пайплайна идёт одновременно.

Single Runner

Проблема. Пайплайн sessionization считает данные инкрементально. Каждый запуск зависит от результата предыдущего. Параллельные запуски дадут неверный результат.

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

Ограничьте параллелизм одним. В Airflow это max_active_runs=1. В Azure Data Factory - настройка Concurrency.
Добавьте depends_on_past=True. Тогда задача не стартует, если та же задача прошлого запуска упала.
На уровне кластера есть StepConcurrencyLevel в AWS EMR.
Если оркестратор не умеет этого, используйте Readiness Marker, который ждёт прошлый запуск.

Цена:

Backfilling. Очень медленный. Параллелить нельзя, это требование бизнеса. Можно лишь не пересчитывать лишние шаги (загрузка без обработки).
Latency. Если запуск раз в час стал занимать 1,5 часа вместо 30 минут, отставание растёт. Страдают все потребители.
В Data Factory лишние запуски не отбрасываются, а встают в очередь (до 100). При переполнении будет ошибка 429.

Concurrent Runner

Проблема. Команда загрузки берёт независимые датасеты из внешних источников. Единственный запуск замедляет всё. Хочется ускорить.

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

Задайте параллелизм больше 1. В Airflow это max_active_runs=5.
Оркестратор берёт следующий запуск, пока число активных ниже лимита.
depends_on_past можно оставить False.

Цена:

Resource starvation. В общем кластере один backfill может занять все слоты. Решение: workload management, квоты на группу пользователей. В serverless-оркестраторах лимиты высокие (Step Functions - до 10 000 параллельных).
Shared state. Параллельные запуски на общем состоянии дают недетерминированный результат. Пример - Dynamic Late Data Integrator: backfill может не запуститься вовсе.

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

Термин	Значение
`Data orchestration layer`	Уровень оркестратора. Задачи, расписание, зависимости
`Data processing layer`	Уровень кода обработки. `Spark`, SQL
`Fan-in`	Много веток сходятся в одну задачу
`Fan-out`	Одна задача расходится в несколько веток
`Trigger rule`	Условие запуска задачи по статусу родителей
`Sensor`	Задача, которая ждёт данные или событие
`Backfill`	Пересчёт прошлых периодов
`Scheduling skew`	Ожидание самой медленной ветки
`Partial data`	Результат по неполным входным данным
`Concurrency`	Число запусков одного пайплайна одновременно
`Resource starvation`	Один пайплайн забирает все ресурсы

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

Делите большой DAG на задачи по границам перезапуска. Дорогие шаги должны перезапускаться отдельно.
Между командами договаривайтесь о схеме и расписании. Держите зависимости явными: Readiness Marker или ExternalTaskSensor.
Для fan-in задайте лимит параллелизма. Иначе 24 или больше веток нагрузят кластер.
Если публикуете частичные данные, пишите флаг полноты в таблицу. Потребитель не должен гадать.
Условия if/else выносите в оркестратор. Не прячьте их в код job.
Для инкрементальных пайплайнов ставьте max_active_runs=1 и depends_on_past. Независимым загрузкам разрешайте параллелизм.
При backfill следите за ресурсами других команд. Используйте квоты.

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

Как выбрать границы между задачами в Local Sequencer? Какие три критерия помогают?
Чем data-based Isolated Sequencer отличается от task-based? Что связано сильнее?
Когда нужен Unaligned Fan-In? Как сообщить потребителю, что данные неполные?
Когда нужен Single Runner, а когда Concurrent Runner? Какие риски у каждого?

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