Пайплайны и валидация: контракты на границах, идемпотентная запись и тихо исчезающие строки

Контракты на границах, а не проверки везде

Пайплайн — это стадии, связанные данными, и инженерный вопрос — где проходят границы доверия. Валидируйте на каждой границе — файлы от других команд, выгрузки из API, всё пересекающее линию владения, и собственный выход перед записью — а внутренность держите чистой: фрейм, прошедший входной контракт, внутренние функции принимают на веру. Оборонительные проверки, рассыпанные по каждому трансформу, хоронят настоящий контракт в шуме и всё равно пропускают вход — а сюрпризы рождаются именно там (Хук был изменением выше по потоку, а не локальным багом).

Инструмент — схема-как-код. pandera выражает контракты DataFrame декларативно — типы колонок, nullable, диапазоны, уникальность — и выдаёт читаемый отчёт об отказе вместо NaN, проползающего четыре стадии до того, как стать видимым:

import pandera as pa

orders_schema = pa.DataFrameSchema(
    {
        "order_id": pa.Column(int, unique=True, nullable=False),
        "currency": pa.Column(str, nullable=False),     # Хук, одна строка
        "amount": pa.Column(float, pa.Check.ge(0)),
    },
    strict=True,            # неожиданные колонки — ошибка, а не пожатие плечами
)

orders = orders_schema.validate(orders, lazy=True)      # lazy: ВСЕ нарушения одним отчётом

lazy=True важен операционно: он собирает все нарушения в один отчёт, и дежурный видит «currency: 12,1% null, order_id: 3 дубликата» в одном упавшем прогоне вместо починки отказов по одному на перезапуск. pydantic играет ту же роль для построчных и конфигурационных данных на границах API; pandera (умеющая и polars) — DataFrame-нативный вариант. Принцип в обоих случаях один: контракт — это код, версионируемый рядом с пайплайном, падающий быстро и с именами — а не assert тремя джойнами ниже.

Идемпотентность: перезапуск не должен дублировать

Спросите себя: что произойдёт, если эта стадия выполнится дважды? Если честный ответ — «выход удвоится», перед вами бомба замедленного действия, взведённая до ближайшего ретрая планировщика или сетевого сбоя.

Ретраи — не исключение, а норма: планировщики перезапускают, поды умирают посреди записи, люди перегоняют вчерашний день. Дисциплина: перезапуск стадии для того же окна входных данных должен дать то же состояние, а не больше строк. Слепой append — антипаттерн; две безопасные формы записи — перезапись партиции, когда выход ключуется окном обработки и заменяется целиком при перезапуске:

out = f"s3://lake/revenue/dt={ds}/"        # ds = обрабатываемый день
result.write_parquet(out)                  # перезапуск заменяет партицию, а не добавляет к ней

— или merge по бизнес-ключу (upsert), когда сток — таблица. Та же структура «ключ-по-окну» даёт инкрементальную обработку: watermark запоминает последнее полностью обработанное окно, каждый прогон берёт только новые партиции, и бэкфилл перестаёт быть аварией — это тот же код, вызванный с явным историческим диапазоном дат, а не правленный руками скрипт с зашитым now(). Лакмус всего дизайна — один вопрос: что случится, если каждая стадия выполнится дважды?

Наблюдаемость: счётчики строк — самая дешёвая сигнализация из возможных

Логируйте rows-in и rows-out для каждой стадии каждого прогона числами, которые машина может сравнить, — и Хук превращается в алерт первого дня: обогащение получило 1 204 000 строк и выдало 1 060 000, падение на 12% при исторической дельте около нуля. Распространите ту же привычку на метрики качества данных как временные ряды — доля null в ключевых колонках, число уникальных, min/max сумм — и вы поймаете отказы, не трогающие счётчики строк: задвоенная запись в справочнике, удваивающая выручку, смена единиц, сдвигающая каждую сумму в 100 раз. Джойны заслуживают одного собственного явного утверждения, потому что именно в джойне живут сюрпризы кардинальности: merge(..., validate="many_to_one") заставляет pandas упасть в момент, когда сторона справочника перестаёт быть уникальной, — взрыв many-to-many пойман на джойне, а не в итогах. И таймстемпы: джойн между timezone-naive и tz-aware колонками — в лучшем случае жёсткая ошибка dtype, в худшем — обе naive, одна в UTC, другая в локальном времени — джойн, почти ничего не совпадающий в часы расхождения смещений, без шанса на ошибку, потому что dtype совпадают.

Оркестрация честно

Ежедневному джобу из трёх стадий нужны cron, makefile и дисциплины выше — заводить ради него workflow-платформу значит платить сложностью без выгоды. Airflow, Dagster или Prefect окупают свою сложность, когда у вас действительно есть DAG зависимых джобов, ретраи по стадиям с backoff, бэкфиллы по диапазонам дат как управляемая операция и команда, которой нужна история прогонов в одном месте. Честная рамка: оркестраторы планируют и показывают ваши стадии; они ничего не валидируют, не дедуплицируют и не версионируют. Контракты, идемпотентность и счётчики — ваш код в любом случае: пайплайн, неправильный под cron, ровно так же неправилен под Airflow, только с дашбордами получше.