Выгрузка свежая, но неполная: как проверять комплект дневных партиций

На странице статуса написано: выгрузка за вчера обновлена десять минут назад. Финансовый отчёт показывает подозрительно низкую выручку. В каталоге действительно есть свежие файлы, но только для двух магазинов из трёх. Проверка max(updated_at) ответила на вопрос о возрасте последнего файла и ничего не сказала о комплектности дня.

У свежести и полноты разные условия успеха. Чтобы проверить полный дневной снимок, заранее определите ожидаемые партиции и сравните их с полученными. Затем убедитесь, что каждая часть относится к нужной дате и принятой версии выгрузки. Самая свежая часть не должна прикрывать отсутствующую соседнюю.

Откуда взять ожидаемый список

Обычно он следует из справочника магазинов, регионов или источников, действовавших на дату данных. Используйте снимок этого справочника. Сегодняшнее закрытие магазина не отменяет его обязательную выгрузку за вчера. Если для закрытого магазина положен нулевой результат, пусть источник отправит явную пустую партицию с количеством ноль и статусом завершения. Отсутствие файла не позволяет отличить ноль продаж от падения задачи.

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

Найдём потерянную часть и повтор

В примере ключ партиции состоит из дня и магазина. Источник дважды прислал A и один раз B. Число полученных файлов равно ожидаемым трём, но полный комплект не собран. Counter позволяет обнаружить повтор до преобразования ключей в множество. Для готовности нужны одновременно отсутствие потерь, лишних ключей и дублей.

from collections import Counter
day = "2026-10-28"
expected = {(day, shop) for shop in ["A", "B", "C"]}
received = [(day, "A"), (day, "A"), (day, "B")]
def inspect(keys):
    counts = Counter(keys)
    missing = sorted(expected - set(counts))
    extra = sorted(set(counts) - expected)
    duplicates = sorted(k for k, n in counts.items() if n > 1)
    return {"missing": missing, "extra": extra, "duplicates": duplicates,
            "ready": not (missing or extra or duplicates)}
first = inspect(received)
print(first)
assert first["missing"] == [(day, "C")]
assert first["duplicates"] == [(day, "A")]
assert not first["ready"]
fixed = inspect([(day, "A"), (day, "B"), (day, "C")])
print(fixed)
assert fixed["ready"]

Проверка сразу объясняет, почему три файла не означают три магазина. Исправленный набор проходит по ключам. После этого нужны проверки содержимого: корректный день внутри строк, уникальность бизнес-ключа и разумные контрольные суммы. Если файл C существует, но содержит данные B, один список имён этого не обнаружит.

Как не смешать две попытки

После повторной выгрузки A может стать новым, а B остаться от старой попытки. Для отчёта важен согласованный snapshot_id. Передавайте его в метаданных каждой части и сравнивайте при приёмке. Если источник разрешает независимое обновление магазинов, определите явное правило согласования: например, по версии закрытия дня. Требование одной попытки подходит не всем источникам, но отсутствие любого правила почти неизбежно смешивает состояния.

До завершения проверки держите новый набор отдельно от активного отчёта. Читателю показывайте последний принятый снимок и его дату. Когда полный комплект принят, переключайте отчёт на его идентификатор согласованной операцией хранилища. Оператору при этом нужны ожидаемые и полученные ключи, а не только красный статус «проверка полноты не прошла».

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

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

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

В курсе есть хранение в Parquet и lakehouse, зависимости Airflow, проверки свежести и сценарий backfill. Следующая задача — включить приёмку дневного комплекта в конец конвейера, чтобы обновление отчёта зависело от результата проверки, а не от появления первого файла. Откройте программу платного курса «Data Engineering Fast‑Track: Kafka → Airflow → Spark».