Set path dynamic for Flink Checkpoint for Flink Job in Yarn Cluser

Question

I am using Yarn for running Flink jobs. For each Flink job I am creating a checkpoint.

I submit a Flink job that runs in my Yarn cluster. I have a polling job which checks if a job has failed on Yarn and restarts it. When the job is submitted again Yarn creates a new application_id for this Flink job. How can I configure the re-submitted Flink job to use the checkpoint for the Flink job that was restarted.

I have set the conf state.savepoints.dir = hdfs://localhost:9000/checkpoint/ in flink-conf.yaml`

When creating the Flink job, streamExecutionEnvironment.setStateBackend(new FsStateBackend("hdfs://localhost:9000/checkpoint/uuid-job-1"));

When I made this setting the checkpoint is saved in path specified in conf file (hdfs://localhost:9000/checkpoint/) and not in the path I set when creating the Flink job.

Any help will be greatly appreciated. Thanks!

Answer 1

Unfortunately you cannot start a new job with checkpoints from old one. What you can do is to use externalized checkpoints. One drawback in flink <=1.5 is that metadata for externalized checkpoints is stored in a single directory for all jobs which is set by the config parameter: state.checkpoints.dir. But you can alter it before each submission.

One additional note from mailing-list thread:

The good news is that Flink 1.5 will rework how externalized checkpoints work a bit: basically, all checkpoints can now be considered externalized and the metadata will be stored in the root directory of the checkpoint, not in one global directory for all jobs. This way, the metadata for externalized checkpoints resides in the checkpoint directory of each job and it should be reasonably simple to restore from that.