Половина batch-прогнозов готова: как публиковать только целый результат

Ночной расчёт должен подготовить прогноз продаж для всех магазинов. Две части уже записаны, третья упала. Аналитик открывает таблицу утром и видит новые значения по части сети, а по остальным магазинам — вчерашние. Каждый отдельный прогноз корректен, но общий план закупки построен из разных дней. Успешная запись строк ещё не означает готовность результата для читателя.

Нужно разделить вычисление и публикацию. Рабочие части можно сохранять по мере готовности. Доступным для отчёта становится только набор, который прошёл проверку полноты и принадлежит одному запуску. Для этого пригодятся уникальный run_id, закрытый список ожидаемых объектов и отдельное состояние опубликованной версии.

Сначала договоримся, что значит «весь результат»

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

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

Читатель продолжает видеть предыдущий полный набор

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

expected = {"A", "B", "C"}
active = {"run": "yesterday", "rows": {"A": 4, "B": 5, "C": 6}}
def commit(run, parts):
    rows = [item for part in parts for item in part]
    ids = [key for key, value in rows]
    if len(ids) != len(set(ids)):
        raise ValueError("duplicate id")
    if set(ids) != expected:
        raise ValueError("incomplete result")
    return {"run": run, "rows": dict(rows)}
try:
    candidate = commit("today", [[("A", 7)], [("B", 8)]])
except ValueError as exc:
    print("blocked:", exc)
assert active["run"] == "yesterday"
active = commit("today", [[("A", 7)], [("B", 8), ("C", 9)]])
print(active)
assert sum(active["rows"].values()) == 24

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

Перенесём правило в хранилище

В базе удобно записывать строки под run_id, затем проверять их и менять ссылку на активный запуск в транзакции. Читатель сначала получает активный run_id и использует его во всём запросе. Если каждый кусок отчёта заново читает указатель, переключение посреди построения страницы снова смешает версии. Для объектного хранилища понадобится его собственный механизм согласованной смены указателя или таблица метаданных с условным обновлением.

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

Старые полные наборы полезно хранить до окончания периода проверки: тогда откат означает возврат ссылки на предыдущий run_id. Рабочие файлы незавершённых попыток очищайте отдельно и только после проверки, что на них никто не ссылается. Это упрощает расследование: по идентификатору запуска можно восстановить вход и причину остановки.

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

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

В курсе вы соберёте batch-инференс, автоматические проверки и рабочий контур вокруг модели. После локальной проверки полноты следующий полезный артефакт — сервисный сценарий, в котором расчёт, API и мониторинг согласованы по версии результата. Откройте программу платного курса «Data Scientist: от данных до деплоя моделей».

Как возобновить вычисление после сбоя.