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

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

news 2025/6/30 17:21:41

在国内说所有可能有些绝对，因为确实有少数大厂技术底蕴确实没的说能做出自己的东西，但其他的至少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);/*这里注释是写一个伪代码，意在你可以处理出任务运行配置，比如内存资源等，传递给sparkLauncherfor (Map.Entry<String, String> conf : otherConfigParams.entrySet()) {sparkLauncher.setConf(conf.getKey(), conf.getValue());}*//*这里注释是写一个伪代码，意在你可以处理出任务主类入参后传递给sparkLauncherif (mainParams.length != 0) {launcher.addAppArgs(mainParams);}*///同步时，必须使用CountDownLatch 不然监听程序是异步的，拿不到任务权柄CountDownLatch countDownLatch = new CountDownLatch(1);// 启动应用并获取任务权柄,并传入一个监听类，监听任务不同状态时的事件SparkAppHandle sparkAppHandle = sparkLauncher.startApplication(new SparkAppHandle.Listener() {//任务运行状态改变的时候触发的操作@Overridepublic 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();}}//任务的上下文发生变动时的事件，一般不用@Overridepublic 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();}}