各位老师Flink中我在创建yarn客户端yarn.init和start方法都成功了，然后运行fli[阿里云实时计算 Flink版]

各位老师Flink中我在创建yarn客户端yarn.init和start方法都成功了，然后运行flink任务也成功了，集群中没有生成新的appid是怎么回事，使用的是yarn perjob模式？private void internalSubmitJob(int step, String jobSteps, Long jobId) {

    Configuration flinkConfiguration = FlinkCommonConfiguration.getCommonFlinkConf();
    String configurationDirectory = FlinkCommonConfiguration.getConfigurationDirectory();

    //创建yarn客户点并初始化、启动
    org.apache.hadoop.conf.Configuration configuration = new org.apache.hadoop.conf.Configuration();
    //指定yarn-site.xml路径
    configuration.addResource(new Path("/xxxxxx/yarn-site.xml"));
    configuration.addResource(new Path("/xxxxxx/core-site.xml"));
    configuration.addResource(new Path("/xxxxxx/resource-types.xml"));
    configuration.addResource(new Path("/xxxxxx/mapred-site.xml"));

// String keytabPath = getClass().getClassLoader().getResource(“appuser.keytab”).getPath();
// String principal = “appuser”
//
// UserGroupInformation.loginUserFromKeytab(“”);

    YarnClient yarnClient = YarnClient.createYarnClient();
    YarnConfiguration yarnConfiguration = new YarnConfiguration(configuration);

    yarnClient.init(yarnConfiguration);
    yarnClient.start();

    // 创建YarnClusterDescriptor
    YarnClusterInformationRetriever clusterInformationRetriever = YarnClientYarnClusterInformationRetriever.create(yarnClient);

    // 集群模式设置为per-job模式
    flinkConfiguration.set(
            DeploymentOptions.TARGET,
            YarnDeploymentTarget.PER_JOB.getName());


    // yarn application name 设置为flink-application
    flinkConfiguration.set(YarnConfigOptions.APPLICATION_NAME, "flink-application");

    YarnLogConfigUtil.setLogConfigFileInConfig(flinkConfiguration, configurationDirectory);

    ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
            .createClusterSpecification();

    YarnClusterDescriptor yarnClusterDescriptor = new YarnClusterDescriptor(flinkConfiguration, yarnConfiguration, yarnClient, clusterInformationRetriever, true);

    StreamGraph streamGraph;
    JobGraph jobGraph;

    // 替换为你的Flink作业jar文件路径
    try {
        streamGraph = FlinkJobRunner.execute(jobId);
        jobGraph = streamGraph.getJobGraph();
        //per job
        ClusterClientProvider clusterClientProvider = yarnClusterDescriptor.deployJobCluster(clusterSpecification,jobGraph, true);
        ClusterClient clusterClient = clusterClientProvider.getClusterClient();
        ApplicationId applicationId = clusterClient.getClusterId();
        String webInterfaceURL = clusterClient.getWebInterfaceURL();
        jobLifeCycleManager.launchNewJobLifeCycle(jobHandler.getJobDescriptor(), jobSteps);
        jobLifeCycleManager.setApplicationId(applicationId.toString());
        jobLifeCycleManager.setClusterClient(clusterClient);
    } catch (Exception e){

// jobLifeCycleManager.raiseException();
log.warn(“sumbit异常”);
throw new RuntimeException(e);
}

}这个是我具体实现方法