JAVA代码实现编程式提交Spark任务

谁践踏了优雅 2023-03-02 13:18 65阅读 0赞

# 三种方法作为记录： #

1）直接调用SparkSubmit的main方法

2）SparkLauncher类的launch方法或者startApplication方法

3）使用RestSubmissionClient的run方法

## SparkSubmit提交任务 ##

String[] param = {
    				"--class", "org.apache.spark.examples.SparkPi",
    				"--master", "local[2]",
    				"/bigdata/spark-2.4.6-bin-hadoop2.6/examples/jars/spark-examples_2.11-2.4.6.jar",
    		};
    
    		SparkSubmit.main(param);
    
    		System.out.println("任务运行完成。。。。。。。。。");

可见，基本上就是把spark-submit脚本放到了java中执行！

**但是要注意：（**Rest URL端口是6066**）**

"--master", "spark://192.168.0.181:6066"

## SparkLauncher提交任务 ##

public static void main(String[] args) throws Exception {
            HashMap<String, String> envParams = new HashMap<>();
            envParams.put("YARN_CONF_DIR", "/home/hadoop/cluster/hadoop-release/etc/hadoop");
            envParams.put("HADOOP_CONF_DIR", "/home/hadoop/cluster/hadoop-release/etc/hadoop");
            envParams.put("SPARK_HOME", "/home/hadoop/cluster/spark-new");
            envParams.put("SPARK_PRINT_LAUNCH_COMMAND", "1");
    
            SparkAppHandle spark = new SparkLauncher(envParams)
                    .setAppResource("/home/hadoop/cluster/spark-new/examples/jars/spark-examples_2.11-2.2.1.jar")
                    .setMainClass("org.apache.spark.examples.SparkPi")
                    .setMaster("yarn")
                    .startApplication();
    
            Thread.sleep(100000);
        }

## RestSubmissionClient的run方法提交 ##

@Test
        public void submit() {
            String appResource = "hdfs://192.168.0.181:8020/opt/guoxiang/wordcount.jar";
            String mainClass = "com.fly.spark.WordCount";
            String[] args = {
                    "hdfs://192.168.0.181:8020/opt/guoxiang/wordcount.txt",
                    "hdfs://192.168.0.181:8020/opt/guoxiang/wordcount"
            };
    
            SparkConf sparkConf = new SparkConf();
            // 下面的是参考任务实时提交的Debug信息编写的
            sparkConf.setMaster("spark://192.168.0.181:6066")
                    .setAppName("carabon" + " " + System.currentTimeMillis())
                    .set("spark.executor.cores", "4")
                    .set("spark.submit.deployMode", "cluster")
                    .set("spark.jars", appResource)
                    .set("spark.executor.memory", "1g")
                    .set("spark.cores.max", "4")
                    .set("spark.driver.supervise", "false");
            Map<String, String> env = System.getenv();
    
            CreateSubmissionResponse response = null;
            try {
                response = (CreateSubmissionResponse)
                        RestSubmissionClient.run(appResource, mainClass, args, sparkConf, new HashMap<>());
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println(response.toJson());
        }