参考：
https://blog.csdn.net/zhangjun5965/article/details/106820591
文章目录
背景
随着flink在流计算领域越来越火，很多公司基于flink搭建了自己的实时计算平台，用户可以在实时平台通过jar或者sql的方式来开发、上线、下线、运维flink任务，避免了构建flink任务的复杂性，使更多不会flink的人能够使用flink。
平时我们自己开发一个flink任务之后，都是通过脚本的方式提交到集群的，但是我们搭建了一个实时计算之后，就不能通过命令行来管理任务了，我们今天就主要讲一下如何通过api的方式来和yarn集群交互。
目前生产环境部署flink任务主要有yarn集群和k8s集群两种方式，虽然说k8s号称下一代资源管理系统，但是对于flink来说，还是有很多不太成熟，所以目前部署flink任务还是以yarn集群为主。
yarn集群部署flink任务目前有两种方式
- yarn session 模式
session模式是在yarn上面预先启动一个集群，然后我们可以将任务部署到集群上，没有任务的时候集群上没有taskmanager，当有了新的任务之后，系统会自动为其分配资源，当任务结束之后，过一段时间（可配置）系统会自动释放资源,这种集群一般适合运行周期比较短的任务，比如批处理任务。
- per job 模式
per job模式是每个任务都启动一个flink集群，这种模式的好处就是资源隔离，不互相影响，任务结束之后，释放相应的资源。这种模式启动任务时间长，一般适合运行常驻任务，比如flink流任务.
案例详解
今天我们主要讲一下如何通过api的方式来停止一个通过per job模式部署在yarn集群上的任务。
命令行停止
我们在命名行模式下可以通过下面的命令来停止一个部署在yarn的per job模式的flink任务.
${FLINK_HOME}/bin/flink stop -m yarn-cluster -yid application_1592386606716_0005 c8ee546129e8480809ee62a4ce7dd91d
我们看到，主要是有两个参数，一个是yarn的applicationId,还有一个是flink的jobId,执行成功之后，会返回一个类似的结果：
Savepoint completed. Path: hdfs://localhost/flink-savepoints/savepoint-c8ee54-ee7a059c2f98
api实现
其实主要的方法就是构造出上面两个id，然后我们使用ClusterClient来停止flink任务.
-
添加配置文件
我们在classpath下添加hadoop和flink的配置文件
构造ApplicationId对象
Configuration flinkConfiguration = new Configuration();
flinkConfiguration.set(YarnConfigOptions.APPLICATION_ID, appId);
YarnClusterClientFactory clusterClientFactory = new YarnClusterClientFactory();
ApplicationId applicationId = clusterClientFactory.getClusterId(flinkConfiguration);
构造jobId
private static JobID parseJobId(String jobIdString) throws CliArgsException{
if (jobIdString == null){
throw new CliArgsException("Missing JobId");
}
final JobID jobId;
try {
jobId = JobID.fromHexString(jobIdString);
} catch (IllegalArgumentException e){
throw new CliArgsException(e.getMessage());
}
return jobId;
}停止任务
通过stopWithSavepoint方法来停止任务,如果savePoint没指定的话，系统将会使用flink配置文件中的state.savepoints.dir选项.
CompletableFuture<String> completableFuture = clusterClient.stopWithSavepoint(
jobID,
true,
savePoint);
String savepoint = completableFuture.get();
System.out.println(savepoint);最后执行完成之后，会返回一个savepoint的地址，和命令行一样，我们可以把这个地址存起来,以便我们后续从这个checkpoint启动。
完整的代码请参考：
https://github.com/zhangjun0x01/bigdata-examples/blob/master/flink/src/main/java/cluster/StopYarnJob.java