如何在Java应用中提交Spark任务?

最近看到有几个Github友关注了Streaming的监控工程——Teddy,所以思来想去还是优化下代码,不能让别人看笑话啊。于是就想改一下之前觉得最丑陋的一个地方——任务提交。

本博客内容基于Spark2.2版本~在阅读文章并想实际操作前,请确保你有:

  1. 一台配置好Spark和yarn的服务器
  2. 支持正常spark-submit --master yarn xxxx的任务提交

老版本

老版本任务提交是基于 ** 启动本地进程,执行脚本spark-submit xxx ** 的方式做的。其中一个关键的问题就是获得提交Spark任务的Application-id,因为这个id是跟任务状态的跟踪有关系的。如果你的资源管理框架用的是yarn,应该知道每个运行的任务都有一个applicaiton_id,这个id的生成规则是:

appplication_时间戳_数字

老版本的spark通过修改SparkConf参数spark.app.id就可以手动指定id,新版本的代码是直接读取的taskBackend中的applicationId()方法,这个方法具体的实现是根据实现类来定的。在yarn中,是通过Yarn的YarnClusterSchedulerBackend实现的,具体的实现逻辑可以参考对应的链接。

感兴趣的同学可以看一下,生成applicaiton_id的逻辑在hadoop-yarn工程的ContainerId中定义。

总结一句话就是,想要自定义id,甭想了!!!!

于是当时脑袋瓜不灵光的我,就想到那就等应用创建好了之后,直接写到数据库里面呗。怎么写呢?

  1. 我事先生成一个自定义的id,当做参数传递到spark应用里面;
  2. 等spark初始化后,就可以通过sparkContext取得对应的application_id以及url
  3. 然后再driver连接数据库,插入一条关联关系

新版本

还是归结于互联网时代的信息大爆炸,我看到群友的聊天,知道了SparkLauncer这个东西,调查后发现他可以基于Java代码自动提交Spark任务。SparkLauncher支持两种模式:

  1. new SparkLauncher().launch() 直接启动一个Process,效果跟以前一样
  2. new SparkLauncher().startApplicaiton(监听器) 返回一个SparkAppHandler,并(可选)传入一个监听器

当然是更倾向于第二种啦,因为好处很多:

  1. 自带输出重定向(Output,Error都有,支持写到文件里面),超级爽的功能
  2. 可以自定义监听器,当信息或者状态变更时,都能进行操作(对我没啥用)
  3. 返回的SparkAppHandler支持 暂停、停止、断连、获得AppId、获得State等多种功能,我就想要这个!!!!

一步一步,代码展示

首先创建一个最基本的Spark程序:

import org.apache.spark.sql.SparkSession;
import java.util.ArrayList;
import java.util.List;

public class HelloWorld {
    public static void main(String[] args) throws InterruptedException {
        SparkSession spark = SparkSession
                .builder()
                //.master("yarn")
                //.appName("hello-wrold")
                //.config("spark.some.config.option", "some-value")
                .getOrCreate();

        List<Person> persons = new ArrayList<>();

        persons.add(new Person("zhangsan", 22, "male"));
        persons.add(new Person("lisi", 25, "male"));
        persons.add(new Person("wangwu", 23, "female"));


        spark.createDataFrame(persons, Person.class).show(false);

        spark.close();

    }
}

然后创建SparkLauncher类:

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

import java.io.IOException;

public class Launcher {
    public static void main(String[] args) throws IOException {
        SparkAppHandle handler = new SparkLauncher()
                .setAppName("hello-world")
                .setSparkHome(args[0])
                .setMaster(args[1])
                .setConf("spark.driver.memory", "2g")
                .setConf("spark.executor.memory", "1g")
                .setConf("spark.executor.cores", "3")
                .setAppResource("/home/xinghailong/launcher/launcher_test.jar")
                .setMainClass("HelloWorld")
                .addAppArgs("I come from Launcher")
                .setDeployMode("cluster")
                .startApplication(new SparkAppHandle.Listener(){
                    @Override
                    public void stateChanged(SparkAppHandle handle) {
                        System.out.println("**********  state  changed  **********");
                    }

                    @Override
                    public void infoChanged(SparkAppHandle handle) {
                        System.out.println("**********  info  changed  **********");
                    }
                });


        while(!"FINISHED".equalsIgnoreCase(handler.getState().toString()) && !"FAILED".equalsIgnoreCase(handler.getState().toString())){
            System.out.println("id    "+handler.getAppId());
            System.out.println("state "+handler.getState());

            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

然后打包工程,打包过程可以参考之前的博客: http://www.cnblogs.com/xing901022/p/7891867.html

打包完成后上传到部署Spark的服务器上。由于SparkLauncher所在的类引用了SparkLauncher,所以还需要把这个jar也上传到服务器上。

[xinghailong@hnode10 launcher]$ ls
launcher_test.jar  spark-launcher_2.11-2.2.0.jar
[xinghailong@hnode10 launcher]$ pwd
/home/xinghailong/launcher

由于SparkLauncher需要指定SPARK_HOME,因此如果你的机器可以执行spark-submit,那么就看一下spark-submit里面,SPARK_HOME是在哪

[xinghailong@hnode10 launcher]$ which spark2-submit
/var/lib/hadoop-hdfs/bin/spark2-submit

最后几行就能看到:

export SPARK2_HOME=/var/lib/hadoop-hdfs/app/spark

# disable randomized hash for string in Python 3.3+
export PYTHONHASHSEED=0

exec "${SPARK2_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@"

综上,我们需要的是:

  1. 一个自定义的Jar,里面包含spark应用和SparkLauncher类
  2. 一个SparkLauncher的jar,spark-launcher_2.11-2.2.0.jar 版本根据你自己的来就行
  3. 一个当前目录的路径
  4. 一个SARK_HOME环境变量指定的目录

然后执行命令启动测试:

java -Djava.ext.dirs=/home/xinghailong/launcher -cp launcher_test.jar Launcher /var/lib/hadoop-hdfs/app/spark yarn

说明:

  1. -Djava.ext.dirs 设置当前目录为java类加载的目录
  2. 传入两个参数,一个是SPARK_HOME;一个是启动模式

观察删除发现成功启动运行了:

id    null
state UNKNOWN
Mar 10, 2018 12:00:52 PM org.apache.spark.launcher.OutputRedirector redirect
INFO: 18/03/10 12:00:52 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
**********  state  changed  **********
...省略一大堆拷贝jar的日志
**********  info  changed  **********
**********  state  changed  **********
Mar 10, 2018 12:00:55 PM org.apache.spark.launcher.OutputRedirector redirect
INFO: 18/03/10 12:00:55 INFO yarn.Client: Application report for application_1518263195995_37615 (state: ACCEPTED)
... 省略一堆重定向的日志
application_1518263195995_37615 (state: ACCEPTED)
id    application_1518263195995_37615
state SUBMITTED
Mar 10, 2018 12:01:00 PM org.apache.spark.launcher.OutputRedirector redirect
INFO: 18/03/10 12:01:00 INFO yarn.Client: Application report for application_1518263195995_37615 (state: RUNNING)
**********  state  changed  **********
... 省略一堆重定向的日志
INFO:    user: hdfs
**********  state  changed  **********
Mar 10, 2018 12:01:08 PM org.apache.spark.launcher.OutputRedirector redirect
INFO: 18/03/10 12:01:08 INFO util.ShutdownHookManager: Shutdown hook called
Mar 10, 2018 12:01:08 PM org.apache.spark.launcher.OutputRedirector redirect
INFO: 18/03/10 12:01:08 INFO util.ShutdownHookManager: Deleting directory /tmp/spark-f07e0213-61fa-4710-90f5-2fd2030e0701

总结

这样就实现了基于Java应用提交Spark任务,并获得其Appliation_id和状态进行定位跟踪的需求了。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏黑泽君的专栏

day32_Hibernate学习笔记_04

  缓存(Cache):是计算机领域非常通用的概念。它介于应用程序和永久性数据存储源(如硬盘上的文件或者数据库)之间,其作用是降低应用程序直接读写硬盘(永久性数...

1002
来自专栏java思维导图

给你一份SpringBoot知识清单

在过去两三年的Spring生态圈,最让人兴奋的莫过于Spring Boot框架。或许从命名上就能看出这个框架的设计初衷:快速的启动Spring应用。因而Spri...

1554
来自专栏LEo的网络日志

python i18n实现

3817
来自专栏技术墨客

Spring核心——数据校验

在Java数据校验详解中详细介绍了Java数据校验相关的功能(简称Bean Validation,涵盖JSR-303、JSR-349、JSR-380),本文将在...

1031
来自专栏IT笔记

SpringBoot开发案例之整合Dubbo消费者

有人卖就有人买,显然是亘古不变的真理,前两篇讲解了SpringBoot+Dubbo的提供者的几种暴露方式,这篇跟大家分享一下消费者如何去订阅属于自己的服务。 相...

3625
来自专栏一枝花算不算浪漫

ActiveMQ的介绍及使用实例.

3687
来自专栏java、Spring、技术分享

记一次unable to create new native thread错误处理过程

unable to create new native thread,看到这里,首先想到的是让运维搞一份线上的线程堆栈(可能通过jstack命令搞定的)。...

1.1K1
来自专栏函数式编程语言及工具

Akka-Cluster(1)- Cluster Singleton 单例节点

关于cluster-singleton我在前面的博文已经介绍过,在这篇我想回顾一下它的作用和使用方法。首先,cluster-singleton就是集群某个节点上...

1183
来自专栏竹清助手

浅谈Linux磁盘修复e2fsck命令

检查 /dev/mapper/VolGroup00-LogVol02 是否有问题,如发现问题便自动修复:

2692
来自专栏Spring相关

第3章—高级装配—配置profile bean

我们正常开发的过程中经常遇到的问题是,开发环境是一套环境,qa测试是一套环境,线上部署又是一套环境。这样从开发到测试再到部署,会对程序中的配置修改多次,尤其是从...

1012

扫码关注云+社区

领取腾讯云代金券