public static JobGraph buildJobGraph(JobParamsInfo jobParamsInfo) throws Exception {
        Properties confProperties = jobParamsInfo.getConfProperties();
        int parallelism = MathUtil.getIntegerVal(confProperties.getProperty(ConfigConstrant.SQL_ENV_PARALLELISM, "1"));
        String flinkConfDir = jobParamsInfo.getFlinkConfDir();

        String[] execArgs = jobParamsInfo.getExecArgs();
        File coreJarFile = new File(findLocalCoreJarPath(jobParamsInfo.getLocalPluginRoot(), jobParamsInfo.getPluginLoadMode()));
        SavepointRestoreSettings savepointRestoreSettings = dealSavepointRestoreSettings(jobParamsInfo.getConfProperties());

        PackagedProgram program = PackagedProgram.newBuilder()
                .setJarFile(coreJarFile)
                .setArguments(execArgs)
                .setSavepointRestoreSettings(savepointRestoreSettings)
                .build();

        Configuration flinkConfig = getFlinkConfiguration(flinkConfDir, confProperties);
        JobGraph jobGraph = PackagedProgramUtils.createJobGraph(program, flinkConfig, parallelism, false);
        return jobGraph;
    }
build JobGraph by JobParamsInfo
Date: 2020/3/4
Company: www.dtstack.com
@author maqi
launcher/src/main/java/com/dtstack/flink/sql/launcher/utils/JobGraphBuildUtil.java