Бюджет конвейера: считайте восстановление вместе с ежедневной загрузкой

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

Начните с обязательства перед потребителем

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

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

Измеряйте производительность на сопоставимом составе данных. Миллион простых записей и миллион записей с тяжёлым соединением не дают одинаковой работы. Для первого планирования удобно считать события в секунду, но в испытании проверьте объём байтов, распределение ключей, число файлов и ограничения чтения источника. Узкое место может оказаться в базе, а свободные работники будут ждать её ответы.

Сколько мощности действительно нужно

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

arrival = 100
outage_seconds = 3 * 3600
backlog = arrival * outage_seconds
recovery_seconds = 3600
required = arrival + backlog / recovery_seconds
print("backlog", backlog, "required_per_second", required)
for capacity in (100, 250, 400):
    net = capacity - arrival
    hours = backlog / net / 3600 if net > 0 else None
    print(capacity, "recovery_hours", hours)
assert backlog == 1080000 and required == 400

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

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

Из мощности получается денежный план

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

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

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

Проверьте план до аварии

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

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

Следующий шаг в практике

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

Связанный разбор: как согласовать backfill с текущей загрузкой.