Может ли оператор flink поддерживать пакетную работу с ApplicationCluster?
когда пакетное задание завершится, каким должно быть состояние ApplicationCluster? Является ли «увеличение restartNonce» разработанным способом повторного запуска задания?
Я пытаюсь использовать оператор flink для развертывания пакетного задания flink и запускать его с помощью cronjob kubernetes каждый день в определенное время.
Ответы 2
Оператор разработан в основном с учетом потоковых заданий, но теоретически пакетные задания также должны работать.
Когда пакетное задание завершается (Flink 1.15 и выше), FlinkDeployment.status.jobStatus.state должен перейти в состояние FINISHED.
Нажатие на restartNonce приведет к повторной отправке задания, если вы установите для upgradeMode значение без сохранения состояния, это запустит его полностью с нуля.
Таким образом, теоретически вы можете использовать cron для запуска restartNonce, но это не тот шаблон, который мы тестировали или использовали в производстве сами.
ApplicationMode не работает, кажется, когда пакетное задание завершено, диспетчер задач перезапускается, и оператор воспринимает это состояние как что-то не так и хочет его восстановить. Итак, я попробовал режим сеанса.
тестовая среда:
- flink-kubernetes-оператор v1.1
- флинк_1_14
операции:
- развернуть сессионный кластер
- развернуть задание сеанса (пакетное задание)
- перезапустите задание сеанса, изменив restartNonce
Результаты:
- первый раз можно запустить сеансовое задание
- при применении конфигурации restartNonce задание сеанса не может запуститься и возникает ошибка «Исключение произошло в обработчике REST: задание не найдено»
2022-11-24 06:20:29,079 — это время, когда должен быть применен restartNonce.
журналы:
2022-11-24 06:14:30,314 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Job Streaming WordCount (24d8e9726de88ab201ea13d48e9cdc8e) switched from state RUNNING to FINISHED.
2022-11-24 06:14:30,314 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Stopping checkpoint coordinator for job 24d8e9726de88ab201ea13d48e9cdc8e.
2022-11-24 06:14:30,315 INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher [] - Job 24d8e9726de88ab201ea13d48e9cdc8e reached terminal state FINISHED.
2022-11-24 06:14:30,317 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Stopping the JobMaster for job 'Streaming WordCount' (24d8e9726de88ab201ea13d48e9cdc8e).
2022-11-24 06:14:30,317 INFO org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore [] - Shutting down
2022-11-24 06:14:30,317 INFO org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPool [] - Releasing slot [5a259aa9f56d090c4c4df02ca2e4f189].
2022-11-24 06:14:30,318 INFO org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPool [] - Releasing slot [7eb2fecceb9aff71e2daa4d358c8031a].
2022-11-24 06:14:30,318 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Close ResourceManager connection abe9ce776ee288f79d2e0a1921fb0896: Stopping JobMaster for job 'Streaming WordCount' (24d8e9726de88ab201ea13d48e9cdc8e).
2022-11-24 06:14:30,318 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Disconnect job manager [email protected]://[email protected]:6123/user/rpc/jobmanager_4 for job 24d8e9726de88ab201ea13d48e9cdc8e from the resource manager.
2022-11-24 06:15:26,189 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Stopping worker gix-flink-cluster-taskmanager-1-3.
2022-11-24 06:15:26,189 INFO org.apache.flink.kubernetes.KubernetesResourceManagerDriver [] - Stopping TaskManager pod gix-flink-cluster-taskmanager-1-3.
2022-11-24 06:15:26,189 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Closing TaskExecutor connection gix-flink-cluster-taskmanager-1-3 because: TaskExecutor exceeded the idle timeout.
2022-11-24 06:15:26,204 WARN org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Discard registration from TaskExecutor gix-flink-cluster-taskmanager-1-3 at (akka.tcp://[email protected]:6122/user/rpc/taskmanager_0) because the framework did not recognize it
2022-11-24 06:15:26,626 WARN akka.remote.ReliableDeliverySupervisor [] - Association with remote system [akka.tcp://[email protected]:6122] has failed, address is now gated for [50] ms. Reason: [Disassociated]
2022-11-24 06:15:26,626 WARN akka.remote.ReliableDeliverySupervisor [] - Association with remote system [akka.tcp://[email protected]:46779] has failed, address is now gated for [50] ms. Reason: [Disassociated]
2022-11-24 06:20:29,079 ERROR org.apache.flink.runtime.rest.handler.job.JobCancellationHandler [] - Exception occurred in REST handler: Job could not be found.
2022-11-24 06:20:31,111 ERROR org.apache.flink.runtime.rest.handler.job.JobCancellationHandler [] - Exception occurred in REST handler: Job could not be found.
2022-11-24 06:20:33,122 ERROR org.apache.flink.runtime.rest.handler.job.JobCancellationHandler [] - Exception occurred in REST handler: Job could not be found.
2022-11-24 06:20:36,152 ERROR org.apache.flink.runtime.rest.handler.job.JobCancellationHandler [] - Exception occurred in REST handler: Job could not be found.
2022-11-24 06:20:40,663 ERROR org.apache.flink.runtime.rest.handler.job.JobCancellationHandler [] - Exception occurred in REST handler: Job could not be found.
2022-11-24 06:20:47,427 ERROR org.apache.flink.runtime.rest.handler.job.JobCancellationHandler [] - Exception occurred in REST handler: Job could not be found.
То, что я предложил, будет работать только на Flink 1.15 и выше. Flink 1.14 еще не имеет необходимой функции, позволяющей сохранять диспетчер заданий после завершения задания в режиме приложения. Из-за этого оператор фактически никогда не видит, что задание завершено, а только то, что кластер исчез. Но это может указывать как на завершение, так и на провал.
большое спасибо за ответ Гьюла. действительно существует некоторая проблема при использовании 'restartNoce'. Поскольку я не могу сообщить о проблеме здесь, github.com/apache/flink-kubernetes-operator, я резюмирую проблему ниже.