当前位置：首页 > article >正文

如何用代码提交spark任务并且获取任务权柄

article 2025/1/8 13:47:28

在国内说所有可能有些绝对，因为确实有少数大厂技术底蕴确实没的说能做出自己的东西，但其他的至少95%数据中台平台研发方案，都是集群中有一个持久化的程序，来接收任务信息，并向集群提交任务同时获取任务的权柄，把任务的appid和日志通过套接字的方式向外提供。

对于spark任务来说无非就是两种形式，要不传过来的是个jar包，要不就是一个sql语句，其他的就是一些任务参数，整体上就和正常的web项目开发没太大差别，不同的就是服务端是以哪种方式处理任务的提交的，给大家分享我经历过的项目中用过的三种处理方式，当然这不是全部，业内确实有真东西，不过人家不开源罢了。

第一种：spark官方提供的SparkLauncherAPI

这种方式，你可以在b站上常见，但是它使用起来限制特别大，我也只用过一次，而且还是在尝试阶段就被放弃了，感觉就和一个半成品一样，最让人难受的一点是它监听任务的最终状态是4个独立的不可变枚举值，而监听程序会终止在第一个触发到的不可变枚举值，就是说如果任务先进入了完成状态，但它的最终状态是失败，那权柄只能生效到完成阶段，后面就监听不到了，观察过源码到时找到了底层更新状态的依据，但是属于受保护包下，不能被public直接调用，还有很多其他的坑，所以说像个半成品

使用它，首先导入SparkLauncherAPI的依赖

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-launcher_2.11</artifactId> <!-- 这里要替换为你的 scala 同版本 ，spark-launcher版本不同，支持的scala也不同具体去maven官方仓库中看-->
    <version>2.1.1</version> <!--SparkLauncherAPI要和你用的 Spark 同版本 -->
</dependency>

随后它的提交任务代码如下

package com.wy;

import org.apache.spark.launcher.SparkAppHandle;
import org.apache.spark.launcher.SparkLauncher;

import java.io.*;
import java.util.concurrent.CountDownLatch;

public class Main {
    public static void main(String[] args) throws IOException, InterruptedException {
        // 使用SparkLauncher提交任务,将任务需要的环境进行封装
        //这里只是核心使用需要的设置，sparkLauncher还支持其他的方法设置其他的内容需要的自己看
        SparkLauncher sparkLauncher = new SparkLauncher()
                .setSparkHome("D:\\mydevtool\\spark-2.1.1-hadoop2.7")//这个程序最终运行的服务器需要同时存在一个spark的install路径
                .setAppResource("D:\\test\\myscala-1.0.jar") //你的任务jar包
                .setMainClass("test.Test1")//任务主类
                .setMaster("local")//master 或者 yarn
                .setAppName("代码提交")//任务名称
                .addAppArgs("D:\\test\\123.txt")//任务的主类入参，这里是个可变参
                .setVerbose(false);

        /*
        这里注释是写一个伪代码，意在你可以处理出任务运行配置，比如内存资源等，传递给sparkLauncher
        for (Map.Entry<String, String> conf : otherConfigParams.entrySet()) {
            sparkLauncher.setConf(conf.getKey(), conf.getValue());
        }*/

        /*
        这里注释是写一个伪代码，意在你可以处理出任务主类入参后传递给sparkLauncher
        if (mainParams.length != 0) {
            launcher.addAppArgs(mainParams);
        }
         */

		//同步时，必须使用CountDownLatch 不然监听程序是异步的，拿不到任务权柄
        CountDownLatch countDownLatch = new CountDownLatch(1);

        // 启动应用并获取任务权柄,并传入一个监听类，监听任务不同状态时的事件
        SparkAppHandle sparkAppHandle = sparkLauncher.startApplication(new SparkAppHandle.Listener() {
            //任务运行状态改变的时候触发的操作
            @Override
            public void stateChanged(SparkAppHandle sparkAppHandle) {

                //状态发生变更时，此时将任务id拿出来
                if ( sparkAppHandle.getAppId() != null ){
                    System.out.println("任务状态："+sparkAppHandle.getState().toString());
                    System.out.println("任务ID："+sparkAppHandle.getAppId());
                }

                //诸如此类，你可以按照你的需求定义任务不同状态下要干的事
                if (sparkAppHandle.getState().compareTo(SparkAppHandle.State.RUNNING)==0){
                    System.out.println("任务开始运行");
                }

                //诸如此类，你可以按照你的需求定义任务不同状态下要干的事
                if (sparkAppHandle.getState().compareTo(SparkAppHandle.State.FINISHED)==0){
                    System.out.println("任务正常完成");
                    countDownLatch.countDown();
                }

                if (sparkAppHandle.getState().compareTo(SparkAppHandle.State.FAILED)==0){
                    System.out.println("任务发生错误");
                    countDownLatch.countDown();
                }

                if (sparkAppHandle.getState().compareTo(SparkAppHandle.State.KILLED)==0){
                    System.out.println("任务被终止");
                    countDownLatch.countDown();
                }

            }

            //任务的上下文发生变动时的事件，一般不用
            @Override
            public void infoChanged(SparkAppHandle sparkAppHandle) {

            }
        });

        //这里的日志的代码，但是输出不能放在这里，按照整体来讲应该要有一个阻塞方法任务开始运行获取流，它实际使用起来发现只有放在监听里面才能正常获取日志，但是开头也说了，监听的生命周期有问题，这也是最终放弃使用的原因之一，你要是只想体验一下，把这部分代码放在监听中在开始run的状态下开始输出就行
        BufferedReader reader = null;
        try{
            String line;
            reader = new BufferedReader(new InputStreamReader(sparkLauncher.launch().getInputStream(),"UTF-8"));
            while ((line = reader.readLine()) != null) {
                System.out.println("日志流在输出："+line);
            }
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {

        }

        countDownLatch.await();
    }
}

第二种：Runtime直接启动脚本

这种方式是最简单的，也是大部分中小项目用的方式，和第一种方式一样的是程序运行在集群中，将submit的日志直返回，使用方通过判断的方式处理出任务的appid和url

public static void main(String[] args) {
		
		String command = "spark-submit --master yarn .......";
        Process p = null;
        String line = null;
        try {
            p = Runtime.getRuntime().exec(command);
            BufferedReader br = new BufferedReader(new InputStreamReader(p.getInputStream()));
            while ((line = br.readLine()) != null){
				//这里将日志输出出去
			}
            br.close();
        } catch (IOException e) {
       	 	e.printStackTrace();
        }
        
	}