背景

• 随着flink在流计算领域越来越火，很多公司基于flink搭建了自己的实时计算平台，用户可以在实时平台通过jar或者sql的方式来开发、上线、下线、运维flink任务，避免了构建flink任务的复杂性，使更多不会flink的人能够使用flink。
• 平时我们自己开发一个flink任务之后，都是通过脚本的方式提交到集群的，但是我们搭建了一个实时计算之后，就不能通过命令行来管理任务了，我们今天就主要讲一下如何通过api的方式来和yarn集群交互。
• 目前生产环境部署flink任务主要有yarn集群和k8s集群两种方式，虽然说k8s号称下一代资源管理系统，但是对于flink来说，还是有很多不太成熟，所以目前部署flink任务还是以yarn集群为主。
• yarn集群部署flink任务目前有两种方式

1. yarn session 模式 session模式是在yarn上面预先启动一个集群，然后我们可以将任务部署到集群上，没有任务的时候集群上没有taskmanager，当有了新的任务之后，系统会自动为其分配资源，当任务结束之后，过一段时间（可配置）系统会自动释放资源,这种集群一般适合运行周期比较短的任务，比如批处理任务。
2. 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