参考：

https://blog.csdn.net/zhangjun5965/article/details/106820591

文章目录

• 背景
• 案例详解
• • 命令行停止
  • api实现

背景
随着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

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

标签：实战,flink,yarn,jobId,任务,api,集群
来源： https://blog.csdn.net/CarloPan/article/details/117844246