Checkpoint Spark Structured Streaming: когда нельзя начать с чистого листа

Поток Spark читал Kafka, обновлял агрегат по заказам и записывал результат в таблицу. После релиза приложение стартует с новым пустым checkpoint. Запуск выглядит здоровым, но часть сообщений читается заново или пропускается — в зависимости от позиции нового источника. Пустой каталог означает новую историю обработки, а не «починили старый поток».

Structured Streaming сохраняет в checkpoint информацию о ходе обработки и, для stateful-запросов, состояние агрегатов. При обычном перезапуске тот же запрос с тем же checkpoint может восстановить прогресс. Если удалить каталог, Spark уже не узнает, какие micro-batch были завершены и какое состояние соответствовало их входным данным. Для оконных сумм это может значить и потерю накопленного состояния, и повторную запись результата.

Три вопроса перед перезапуском

Сначала определите, остался ли исходный checkpoint доступен и цел. Затем проверьте, не изменился ли запрос: число и тип источников, схема состояния агрегата, ключи группировки и конфигурация state store. Наконец, выясните, может ли sink распознать повтор той же логической записи. Без ответов на эти вопросы переключать путь checkpoint опасно даже при зелёном статусе приложения.

Документация Spark 4.1 отдельно предупреждает: изменение числа или типа источников при возобновлении из того же checkpoint не поддерживается. Для stateful-операций смена схемы состояния тоже не разрешается как обычный рестарт. Например, если раньше агрегировали продажи по shop_id, а теперь по shop_id, region, старые байты состояния не становятся новой структурой автоматически. Нужна отдельная миграция или новая обработка с контролируемым backfill.

Новый checkpoint — это новая обработка

Если несовместимое изменение всё-таки требует нового checkpoint, зафиксируйте точку переключения. При чтении Kafka выберите начальные позиции для нового запроса и сопоставьте их с уже записанными результатами. Если начать раньше, sink должен выдерживать повтор; если позже — появится дыра. Для файлового источника и оконного агрегата дополнительно решите, откуда новый поток восстановит нужную историю состояния.

Не рассчитывайте, что внешний sink всегда даст exactly-once только из-за checkpoint. Гарантия зависит от источника, sink и способа записи. Например, произвольный внешний API внутри foreachBatch может принять запрос, после чего driver упадёт до отметки успешного micro-batch. Следующий запуск повторит batch. Используйте ID batch и устойчивый бизнес-ключ для идемпотентного сохранения результата.

В эксплуатации храните checkpoint в надёжном постоянном хранилище, отдельно от временных файлов приложения. У одного активного запроса должен быть свой путь; совместное использование каталога двумя независимыми query смешивает их историю. Проверяйте права доступа и сохранность каталога при развёртывании так же внимательно, как доступность входной темы.

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

Разделяйте обычный рестарт и изменение логики. Сбой executor при неизменном запросе — типичный случай восстановления из прежнего checkpoint. Замена ключа группировки или источника — миграция состояния. Её нельзя оформить удалением каталога и считать тем же потоком. Назовите новую версию расчёта, перепроверьте начальную позицию и сопоставьте результаты на переходном интервале.

Сценарий для проверки: поток обработал batch 120, записал результат во внешнюю таблицу, но упал до надёжной фиксации завершения. После запуска из checkpoint batch может быть повторён. Если sink поддерживает атомарную фиксацию по batch ID или merge по устойчивому ключу, повтор не меняет логический результат. Если код делает обычный append, итог может удвоиться. Номер batch полезен для технической диагностики, а бизнес-ключ нужен для устойчивости при новой истории обработки.

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

На курсе Data Engineering Fast‑Track Kafka, Spark и lakehouse объединяются в проверяемый поток и сценарии восстановления. Откройте программу, чтобы отрабатывать перезапуск вместе с последствиями для конечной таблицы.

Если проблема начинается с недоступной старой позиции Kafka, посмотрите разбор фиксации offset; checkpoint Spark добавляет к позиции ещё состояние запроса.