Flink на практике: состояние, контрольные точки и восстановление

Прямой ответ

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

Базовая схема задания:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(30_000, CheckpointingMode.EXACTLY_ONCE);

env.fromSource(kafkaSource, WatermarkStrategy
        .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(30))
        .withTimestampAssigner((o, ts) -> o.eventTime), "orders")
   .keyBy(Order::shopId)
   .window(TumblingEventTimeWindows.of(Time.minutes(5)))
   .aggregate(new OrderCountAggregate())
   .sinkTo(resultSink);

Состояние хранится в управляемом хранилище Flink, поэтому объём не ограничен памятью задачи.

Контрольная точка выполняется асинхронно: барьер проходит по потоку, и каждая задача сохраняет состояние при получении барьера. Обработка не останавливается на время снимка.

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

Контрольные точки и savepoint: различие назначения

Оба механизма создают снимок состояния, но используются по-разному.

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

Savepoint создаётся по запросу перед изменением задания и сохраняется до явного удаления. Используется при обновлении кода, изменении параллелизма и переносе на другой кластер.

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

Три требования к корректному восстановлению.

Совместимость состояния. Изменение структуры состояния в коде требует миграции, поскольку снимок содержит данные прежней структуры.

Согласованность источника. Источник должен поддерживать восстановление с сохранённого смещения. Источник без такой поддержки приводит к повторной обработке либо к пропуску.

Идемпотентность приёмника. Запись результата во внешнюю систему выполняется вне снимка, поэтому повтор после восстановления создаёт дубль, если приёмник не идемпотентен. О семантике — в статье про гарантии доставки.

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

Оконная агрегация во Flink использует управляемое состояние для накопления значений до закрытия окна; об окнах — в статье про оконную агрегацию. Момент закрытия окна определяется watermark; о порядке событий — в статье про watermark.

Хранилище состояния и его влияние на восстановление

Управляемое состояние размещается в хранилище, выбор которого определяет скорость снимка и время восстановления.

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

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

Практическое следствие: инкрементальные снимки снижают длительность контрольной точки при большом состоянии, поэтому длительность не растёт линейно с объёмом данных.

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

Размер состояния влияет на время восстановления: при запуске задачи состояние загружается из снимка, поэтому крупное состояние увеличивает время до возобновления обработки. Ограничение объёма состояния достигается сокращением размера окна и допустимой задержки.

Практика: эксплуатация задания потоковой обработки

1. Включайте контрольные точки с интервалом по требованию к потере. Короткий интервал снижает объём повторной обработки после сбоя.

2. Создавайте savepoint перед изменением задания. Остановка без снимка лишает возможности восстановить состояние.

3. Проверяйте совместимость состояния при изменении кода. Снимок содержит данные прежней структуры, поэтому изменение требует миграции.

4. Делайте приёмник идемпотентным. Запись результата выполняется вне снимка, поэтому повтор после восстановления создаёт дубль.

5. Следите за длительностью контрольных точек. Превышение интервала означает, что снимки не завершаются и накопление растёт.

6. Мониторьте размер состояния. Рост размера увеличивает длительность снимка и время восстановления.

7. Проверяйте восстановление в тестах. Остановка задачи и восстановление из контрольной точки показывают корректность настройки.

Рекомендации:

1. Различайте восстановление после сбоя и обновление задания. Первое выполняется автоматически из контрольной точки, второе — из savepoint с возможным изменением структуры.
2. Учитывайте стоимость хранения состояния. Хранилище состояния занимает диск и память, поэтому объём планируется заранее.
3. Проверяйте поведение при перебалансировке параллелизма. Изменение числа параллельных задач требует перераспределения состояния.
4. Не смешивайте состояние задания и внешнее хранилище. Состояние в снимке и данные во внешней базе восстанавливаются независимо, поэтому согласованность требует идемпотентности.
5. Документируйте процедуру обновления. Последовательность savepoint — остановка — развёртывание — восстановление фиксируется, поскольку пропуск шага приводит к потере состояния.

Практический сценарий: состояние потеряно после обновления задания

Контекст: после развёртывания новой версии задания агрегаты начали считаться с нуля, хотя обработка не прерывалась.

1. Диагностика. Задание запущено без восстановления из savepoint. Проверка показала, что предыдущая остановка выполнена без создания снимка, поэтому состояние окон и агрегатов отсутствовало.
2. Причина. Задание остановлено принудительно без savepoint. Автоматические контрольные точки хранились в каталоге, очищаемом при перезапуске, поэтому снимок оказался недоступен.
3. Действия. (а) Перевели хранилище контрольных точек на постоянное внешнее хранилище, не очищаемое при перезапуске; (б) добавили шаг создания savepoint в процедуру обновления как обязательный; (в) восстановили агрегаты повторной обработкой потока с сохранённого смещения источника; (г) добавили проверку наличия savepoint перед остановкой задания в конвейер развёртывания; (д) добавили метрику возраста последней успешной контрольной точки для обнаружения остановки снимков.
4. Результат. Обновление задания выполняется с сохранением состояния, а отсутствие снимка обнаруживается до остановки.

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

FAQ

Чем savepoint отличается от контрольной точки? Контрольная точка создаётся автоматически для восстановления после сбоя и удаляется по устаревании. Savepoint создаётся по запросу и хранится до явного удаления.

Нужно ли останавливать обработку для контрольной точки? Нет. Снимок выполняется асинхронно с прохождением барьера по потоку, обработка продолжается.

Что произойдёт при изменении структуры состояния в коде? Восстановление из прежнего снимка требует миграции состояния, поскольку снимок содержит данные прежней структуры.

Достаточно ли exactly-once во Flink для отсутствия дублей? Нет. Запись во внешнюю систему выполняется вне снимка, поэтому приёмник должен быть идемпотентным.

Источники: Apache Flink stateful stream processing, checkpoints, savepoints — ссылки в шапке.