public void exec() throws Exception {

        Preconditions.checkArgument(StringUtils.equalsIgnoreCase(jobParamsInfo.getPluginLoadMode(), EPluginLoadMode.CLASSPATH.name()),
                "standalone only supports classpath mode");

        JobGraph jobGraph = JobGraphBuildUtil.buildJobGraph(jobParamsInfo);
        Configuration flinkConfiguration = JobGraphBuildUtil.getFlinkConfiguration(jobParamsInfo.getFlinkConfDir(), jobParamsInfo.getConfProperties());

        if (!StringUtils.isBlank(jobParamsInfo.getUdfJar())) {
            JobGraphBuildUtil.fillUserJarForJobGraph(jobParamsInfo.getUdfJar(), jobGraph);
        }

        JobGraphBuildUtil.fillJobGraphClassPath(jobGraph);

        ClusterDescriptor clusterDescriptor = StandaloneClientFactory.INSTANCE.createClusterDescriptor("", flinkConfiguration);

        try {
            ClusterClientProvider clusterClientProvider = clusterDescriptor.retrieve(StandaloneClusterId.getInstance());
            ClusterClient clusterClient = clusterClientProvider.getClusterClient();

            JobExecutionResult jobExecutionResult = ClientUtils.submitJob(clusterClient, jobGraph);
            String jobId = jobExecutionResult.getJobID().toString();
            LOG.info("jobID:" + jobId);
        } finally {
            try {
                clusterDescriptor.close();
            } catch (Exception e) {
                LOG.info("Could not properly close the yarn cluster descriptor.", e);
            }
        }

    }
Date: 2020/3/6
Company: www.dtstack.com
@author maqi
launcher/src/main/java/com/dtstack/flink/sql/launcher/executor/StandaloneExecutor.java