public void exec() throws Exception {
        JobGraph jobGraph = JobGraphBuildUtil.buildJobGraph(jobParamsInfo);
        if (!StringUtils.isBlank(jobParamsInfo.getUdfJar())) {
            JobGraphBuildUtil.fillUserJarForJobGraph(jobParamsInfo.getUdfJar(), jobGraph);
        }

        Configuration flinkConfiguration = JobGraphBuildUtil.getFlinkConfiguration(jobParamsInfo.getFlinkConfDir(), jobParamsInfo.getConfProperties());
        appendApplicationConfig(flinkConfiguration, jobParamsInfo);

        YarnClusterDescriptor clusterDescriptor = (YarnClusterDescriptor) YarnClusterClientFactory.INSTANCE
                .createClusterDescriptor(jobParamsInfo.getYarnConfDir(), flinkConfiguration);

        try {
            List<File> shipFiles = getShipFiles(jobParamsInfo.getFlinkJarPath(), jobParamsInfo.getPluginLoadMode(), jobGraph, clusterDescriptor);

            if (jobParamsInfo.getAddShipFile() != null) {
                List<String> addShipFilesPath = parsePathFromStr(jobParamsInfo.getAddShipFile());
                for (String path : addShipFilesPath) {
                    shipFiles.add(getAddShipFile(path));
                }
            }

            dumpSameKeytab(flinkConfiguration, shipFiles);
            clusterDescriptor.addShipFiles(shipFiles);

            ClusterSpecification clusterSpecification = YarnClusterClientFactory.INSTANCE.getClusterSpecification(flinkConfiguration);
            ClusterClientProvider<ApplicationId> applicationIdClusterClientProvider = clusterDescriptor.deployJobCluster(clusterSpecification, jobGraph, true);

            String applicationId = applicationIdClusterClientProvider.getClusterClient().getClusterId().toString();
            String flinkJobId = jobGraph.getJobID().toString();

            LOG.info(String.format("deploy per_job with appId: %s, jobId: %s", applicationId, flinkJobId));
        } finally {
            try {
                clusterDescriptor.close();
            } catch (Exception e) {
                LOG.info("Could not properly close the yarn cluster descriptor.", e);
            }
        }
    }
Date: 2020/3/4
Company: www.dtstack.com
@author maqi
launcher/src/main/java/com/dtstack/flink/sql/launcher/executor/YarnJobClusterExecutor.java