Задать динамический путь для контрольной точки Flink для задания Flink в Yarn Cluser

Question

Я использую пряжу для выполнения заданий Flink.Для каждого задания Flink я создаю контрольную точку.

Я отправляю задание Flink, которое выполняется в моем кластере пряжи.У меня есть опрос, который проверяет, не сработала ли работа на пряже, и перезапускает ее.Когда задание снова отправляется, Yarn создает новый application_id для этого задания Flink.Как настроить повторно отправленное задание Flink для использования контрольной точки для задания Flink, которое было перезапущено.

Я установил conf state.savepoints.dir = hdfs://localhost:9000/checkpoint/ в flink-conf.yaml`

При создании задания Flink streamExecutionEnvironment.setStateBackend(new FsStateBackend("hdfs://localhost:9000/checkpoint/uuid-job-1"));

Когда я сделал эту настройку, контрольная точкасохраняется по пути, указанному в файле conf (hdfs://localhost:9000/checkpoint/), а не по пути, указанному мной при создании задания Flink.

Любая помощь будет принята с благодарностью.Спасибо!

Answer 1

К сожалению, вы не можете начать новую работу с контрольных точек со старой.Что вы можете сделать, это использовать внешние контрольные точки .Один недостаток flink <= 1.5 заключается в том, что метаданные для внешних контрольных точек хранятся в одном каталоге для всех заданий, который задается параметром config: <code>state.checkpoints.dir.Но вы можете изменять его перед каждой отправкой.

Одна дополнительная заметка из списка рассылки thread :

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