技术标签: spark spark launcher Spark
在介绍之前,我先附上spark 官方文档地址:
http://spark.apache.org/docs/latest/api/java/org/apache/spark/launcher/package-summary.html
个人源码github地址:
https://github.com/yyijun/framework/tree/master/framework-spark
spark-submit \
--master yarn \
--deploy-mode cluster \
--driver-memory 4g \
--driver-cores 4 \
--num-executors 20 \
--executor-cores 4 \
--executor-memory 10g \
--class com.yyj.train.spark.launcher.TestSparkLauncher \
--conf spark.yarn.jars=hdfs://hadoop01.xxx.xxx.com:8020/trainsparklauncher/jars/*.jar \
--jars $(ls lib/*.jar| tr '\n' ',') \
lib/ train-spark-1.0.0.jar
--conf spark.yarn.jars:提交算法到yarn集群时算法依赖spark安装包lib目录下的jar包,如果不指定,则每次启动任务都会先上传相关依赖包,耗时严重;
--jars:算法依赖的相关包,spark standalone模式、yarn模式都有用,多个依赖包用逗号”,”分隔;
val spark = SparkSession
.builder
.appName("TestSparkLauncher")
.master("yarn")
.config("deploy.mode", "cluster")
.config("spark.yarn.jars", "hdfs://hadoop01.xxx.xxx.com:8020/trainsparklauncher/jars/*.jar")
.config("spark.sql.warehouse.dir", "/user/hive/warehouse")
.enableHiveSupport()
.getOrCreate()
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-yarn_2.11</artifactId>
<version>2.1.0</version>
</dependency>
1、从大数据平台下载hadoop相关的xml配置文件:
core-site.xml:必须;
hdfs-site.xml:必须;
hive-site.xml:提交的算法里面用到spark on hive时需要此文件;
yarn-site.xml:提交算法到yarn时必须要此文件;
2、准备自己的算法包,这里对应替换为自己的算法包:
train-spark-1.0.0.jar和train-common-1.0.0.jar
3、上传spark安装目录下jars目录下相关的jar包到hdfs:hadoop fs –put –f /opt/cloudera/parcels/SPARK2/lib/spark2/jars /hdfs目录
测试提交算法
package com.yyj.framework.spark.launcher;
import java.io.File;
import java.util.HashMap;
import java.util.Map;
/**
* Created by yangyijun on 2019/5/20.
* 提交spark算法入口类
*/
public class SparkLauncherMain {
public static void main(String[] args) {
System.out.println("starting...");
String confPath = "/Users/yyj/workspace/alg/src/main/resources";
System.out.println("confPath=" + confPath);
//开始构建提交spark时依赖的jars
String rootPath = "/Users/yyj/workspace/alg/lib/";
File file = new File(rootPath);
StringBuilder sb = new StringBuilder();
String[] files = file.list();
for (String s : files) {
if (s.endsWith(".jar")) {
sb.append("hdfs://hadoop01.xxx.xxx.com:8020/user/alg/jars/");
sb.append(s);
sb.append(",");
}
}
String jars = sb.toString();
jars = jars.substring(0, jars.length() - 1);
Map<String, String> conf = new HashMap<>();
conf.put(SparkConfig.DEBUG, "false");
conf.put(SparkConfig.APP_RESOURCE, "hdfs://hadoop01.xxx.xxx.com:8020/user/alg/jars/alg-gs-offline-1.0.0.jar");
conf.put(SparkConfig.MAIN_CLASS, "com.yyj.alg.gs.offline.StartGraphSearchTest");
conf.put(SparkConfig.MASTER, "yarn");
//如果是提交到spark的standalone集群则采用下面的master
//conf.put(SparkConfig.MASTER, "spark://hadoop01.xxx.xxx.com:7077");
conf.put(SparkConfig.APP_NAME, "offline-graph-search");
conf.put(SparkConfig.DEPLOY_MODE, "client");
conf.put(SparkConfig.JARS, jars);
conf.put(SparkConfig.HADOOP_CONF_DIR, confPath);
conf.put(SparkConfig.YARN_CONF_DIR, confPath);
conf.put(SparkConfig.SPARK_HOME, "/Users/yyj/spark2");
conf.put(SparkConfig.DRIVER_MEMORY, "2g");
conf.put(SparkConfig.EXECUTOR_CORES, "2");
conf.put(SparkConfig.EXECUTOR_MEMORY, "2g");
conf.put(SparkConfig.SPARK_YARN_JARS, "hdfs://hadoop01.xxx.xxx.com:8020/user/alg/jars/*.jar");
conf.put(SparkConfig.APP_ARGS, "params");
SparkActionLauncher launcher = new SparkActionLauncher(conf);
boolean result = launcher.waitForCompletion();
System.out.println("============result=" + result);
}
}
构造SparkLauncher对象,配置Spark提交算法相关参数及说明
private SparkLauncher createSparkLauncher() {
logger.info("actionConfig:\n" + JSON.toJSONString(conf, true));
this.debug = Boolean.parseBoolean(conf.get(SparkConfig.DEBUG));
Map<String, String> env = new HashMap<>();
//配置hadoop的xml文件本地路径
env.put(SparkConfig.HADOOP_CONF_DIR, conf.get(SparkConfig.HADOOP_CONF_DIR));
//配置yarn的xml文件本地路径
env.put(SparkConfig.YARN_CONF_DIR, conf.get(SparkConfig.HADOOP_CONF_DIR));
SparkLauncher launcher = new SparkLauncher(env);
//设置算法入口类所在的jar包本地路径
launcher.setAppResource(conf.get(SparkConfig.APP_RESOURCE));
//设置算法入口类保证包名称及类名,例:com.yyj.train.spark.launcher.TestSparkLauncher
launcher.setMainClass(conf.get(SparkConfig.MAIN_CLASS));
//设置集群的master地址:yarn/spark standalone的master地址,例:spark://hadoop01.xxx.xxx.com:7077
launcher.setMaster(conf.get(SparkConfig.MASTER));
//设置部署模式:cluster(集群模式)/client(客户端模式)
launcher.setDeployMode(conf.get(SparkConfig.DEPLOY_MODE));
//设置算法依赖的包的本地路径,多个jar包用逗号","隔开,如果是spark on yarn只需要把核心算法包放这里即可,
// spark相关的依赖包可以预先上传到hdfs并通过 spark.yarn.jars参数指定;
// 如果是spark standalone则需要把所有依赖的jar全部放在这里
launcher.addJar(conf.get(SparkConfig.JARS));
//设置应用的名称
launcher.setAppName(conf.get(SparkConfig.APP_NAME));
//设置spark客户端安装包的home目录,提交算法时需要借助bin目录下的spark-submit脚本
launcher.setSparkHome(conf.get(SparkConfig.SPARK_HOME));
//driver的内存设置
launcher.addSparkArg(SparkConfig.DRIVER_MEMORY, conf.getOrDefault(SparkConfig.DRIVER_MEMORY, "4g"));
//driver的CPU核数设置
launcher.addSparkArg(SparkConfig.DRIVER_CORES, conf.getOrDefault(SparkConfig.DRIVER_CORES, "2"));
//启动executor个数
launcher.addSparkArg(SparkConfig.NUM_EXECUTOR, conf.getOrDefault(SparkConfig.NUM_EXECUTOR, "30"));
//每个executor的CPU核数
launcher.addSparkArg(SparkConfig.EXECUTOR_CORES, conf.getOrDefault(SparkConfig.EXECUTOR_CORES, "4"));
//每个executor的内存大小
launcher.addSparkArg(SparkConfig.EXECUTOR_MEMORY, conf.getOrDefault(SparkConfig.EXECUTOR_MEMORY, "4g"));
String sparkYarnJars = conf.get(SparkConfig.SPARK_YARN_JARS);
if (StringUtils.isNotBlank(sparkYarnJars)) {
//如果是yarn的cluster模式需要通过此参数指定算法所有依赖包在hdfs上的路径
launcher.setConf(SparkConfig.SPARK_YARN_JARS, conf.get(SparkConfig.SPARK_YARN_JARS));
}
//设置算法入口参数
launcher.addAppArgs(new String[]{conf.get(SparkConfig.APP_ARGS)});
return launcher;
}
准spark安装包,用于提交spark算法的客户端,因为提交算法的时候需要用到Spark的home目录下的bin/spark-submit脚本
重命名conf目录下的spark-env.sh脚本,否则会包如下的错误。原因是spark-env.sh里面配置了大数据平台上的路径,而在提交算法的客户端机器没有对应路径
debug模式提交或者非debug模式
/**
* Submit spark application to hadoop cluster and wait for completion.
*
* @return
*/
public boolean waitForCompletion() {
boolean success = false;
try {
SparkLauncher launcher = this.createSparkLauncher();
if (debug) {
Process process = launcher.launch();
// Get Spark driver log
new Thread(new ISRRunnable(process.getErrorStream())).start();
new Thread(new ISRRunnable(process.getInputStream())).start();
int exitCode = process.waitFor();
System.out.println(exitCode);
success = exitCode == 0 ? true : false;
} else {
appMonitor = launcher.setVerbose(true).startApplication();
success = applicationMonitor();
}
} catch (Exception e) {
logger.error(e);
}
return success;
}
非debug模式提交时,控制台获取处理结果信息
///
// private functions
///
private boolean applicationMonitor() {
appMonitor.addListener(new SparkAppHandle.Listener() {
@Override
public void stateChanged(SparkAppHandle handle) {
logger.info("****************************");
logger.info("State Changed [state={0}]", handle.getState());
logger.info("AppId={0}", handle.getAppId());
}
@Override
public void infoChanged(SparkAppHandle handle) {
}
});
while (!isCompleted(appMonitor.getState())) {
try {
Thread.sleep(3000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
boolean success = appMonitor.getState() == SparkAppHandle.State.FINISHED;
return success;
}
private boolean isCompleted(SparkAppHandle.State state) {
switch (state) {
case FINISHED:
return true;
case FAILED:
return true;
case KILLED:
return true;
case LOST:
return true;
}
return false;
}
可以从处理结果中获取到app ID,用于杀掉yarn任务时使用
//访问URL:
http://<rm http address:port>/ws/v1/cluster/apps/{appID}
//例子
http://localhost:8088/ws/v1/cluster/apps/application15617064805542301
访问详情地址,返回数据格式如下:
"id": "application15617064805542301",--任务ID
"user": "haizhi",--提交任务的用户名称
"name": "TestSparkLauncher",--应用名称
"queue": "root.users.haizhi",--提交队列
"state": "FINISHED",--任务状态
"finalStatus": "SUCCEEDED",--最终状态
"progress": 100,--任务进度
"trackingUI": "History",
"trackingUrl": "http://hadoop01.xx.xxx.com:18088/proxy/application15617064805542301/A",
"diagnostics":"",--任务出错时的主要错误信息
"clusterId": 1561706480554,
"applicationType": "SPARK",--任务类型
"startedTime": 1562808570464,--任务开始时间,单位毫秒
"finishedTime": 1562808621348,--任务结束时间,单位毫秒
"elapsedTime": 50884,--任务耗时,毫秒
"amContainerLogs": "http://hadoop01.xx.xxx.com:8042/node/containerlogs/container15617064805542301_01_000001/haizhi",--任务详细日志
"amHostHttpAddress": "hadoop01.xx.xxx.com:8042",
"memorySeconds": 198648,--任务分配到的内存数,单位MB
"vcoreSeconds": 145,--任务分配到的CPU核数
"logAggregationStatus": "SUCCEEDED"
请求URL:http://<rm http address:port>/ws/v1/cluster/apps/{appid}/state
请求方式:put
请求参数: { "state": "KILLED" }
例:
请求URL:http://192.168.1.3:18088/ws/v1/cluster/apps/application15617064805542302/state
请求方式:put
请求参数: { "state": "KILLED" }
文章浏览阅读307次。ASP.NET MVC 是一个全新的Web应用框架ASP.NET 代表支撑应用框架的技术平台,表明ASP.NET MVC和传统的WebForm应用框架一样,都是建立在ASP.NET平台之上。MVC 表示该框架背后的设计思想,意味着ASP.NET MVC采用了MVC架构模式。MVC在20世纪70年代后期出现,产生于Xerox PARC施乐公司的帕洛阿尔托研究中心的Small..._aspnet mvvm
文章浏览阅读350次。解压压缩包后,打开终端,进入当前chromedriver所在的路径,将chromedriver移动到默认路径(/usr/local/bin/)二、下载对应版本的浏览器驱动器:找到版本一致或最接近的版本的驱动器。若列表中含有selenium则表明安装成功。记得勾选上继承全局包,要不然会报错。_m1 mac selenium配置
文章浏览阅读4.6k次。谨以此片献给各种遇到坑的朋友。官网上的OpenCV 插件,根据github 上的描述 是针对UE4 4.16 版本以下的版本来提供的 OpenCV 采用的是3.0.0 , 也可以使用openCv 3.2.0环境描述:Ue4 4.15 OpenCV 3.2.0固有插件配置步骤:1, 创建一个C++ 的工程,打开后, 关闭即可。2,将OpenCV 的插件文件 解压缩之后, 将里面的文件..._ue4 opencv操作texture
文章浏览阅读1w次,点赞3次,收藏11次。直接在cmd命令行输入:conda install --channel https://conda.anaconda.org/menpo opencv3 接着根据提示按Y即可
文章浏览阅读646次。#include#includestruct mubang{int x;int y;}a[5005];int cmp(const void *a,const void *b){struct mubang *c=(struct mubang *)a;struct mubang *d=(struct mubang *)b;if(c->x!=d->x)re_nyoj 心急的c小加
文章浏览阅读1.7k次。学习JavaEE的你们是不是一头雾水呢? C/S client/server 客户端/服务器 B/C browser/server 浏览器/服务器 通过浏览器访问到对应页面,发送请求(同步请求,异步请求,ajax请求,会读接口文档)---服务器(云服务器:腾讯云服务器 阿里云服务器等等)---应用服务器Tomcat---匹配servlet---service业务处理---dao层 JDBC Hibemate,ORM---数据库---File---servlet(控制器Controller)MVC_javaee基础知识
文章浏览阅读910次,点赞28次,收藏19次。交互性检索是在检索用户不能构建良好的检索式(关键词)的情况下,通过与检索平台交流互动并不断修改检索式,从而获得较准确检索结果的过程。新闻推荐需要:获取用户请求,召回候选新闻,对候选新闻进行排序,最终给用户推出新闻。将用户持续浏览新闻的推荐过程看成一个决策过程,通过强化学习学习每一次推荐的最佳策略,提高用户的点击率。无人驾驶被认为是强化学习短期内能技术落地的一个应用方向,很多公司投入大量资源在无人驾驶上,其中百度的无人巴士“阿波龙”已经在北京、武汉等地展开试运营,自动无人驾驶的行车视野如下图所示。
文章浏览阅读3.5w次,点赞102次,收藏685次。简历是什么找工作之前投递的个人信息和工作能力----不全面应该是:个人当前阶段的价值体现前者:我能卖多少钱;后者:我现在值多少钱建议:每隔半年/一年给自己做总结的时候写一份简历(相当于个人价值总结)面试要刻意、精心准备公司内部晋升答辩,需要精心准备(ppt、演讲基本要精心准备一个月的时间)面试,简历,同样需要精心准备目录面试官如何分析一份简历简历模板和内容个人信息教育经历专业技能工作经历项目经历体现自己的亮点课程总结注意:不要造假学历造假:学信网可查工作经历造假:可_前端简历
文章浏览阅读2.6k次。https://www.dwhd.org/20160724_085212.html_有gcc高版本的yum库
文章浏览阅读320次。1起步依赖原理分析 1.1分析spring-boot-starter-parent 按住Ctrl点击pom.xml中的spring-boot-starter-parent,跳转到了spring-boot-starter-parent的pom.xml,xml配置如下(只摘抄了部分重点配置):<parent> <groupId>org.sp..._springboot 自动配置分析与整合测试
文章浏览阅读9.1k次。最近想要使用Crypto.Cipher 的AES模块,可总是找不到Crypto.Cipher接下来介绍几个坑1.AES是在pyCrypto中而不是crypto中2.这个pyCrypto中间的C一定要大写,不然也不能用接下来介绍正确的安装姿势1.首先需要现在一款编译器Microsoft Visual C++ Compiler for Python 2.7下载地址https://www.m..._python2.7 使用pycrypto
文章浏览阅读2.9k次,点赞4次,收藏7次。递归算法:将正整数n表示成一系列正整数之和,n=n1+n2+...+nk,其中n1>=n2>=n3>=...>=nk>=1,k>=1。正整数n的这种表示称为正整数n的划分。正整数n的不同的划分个数城外正整数n的划分数,记作p(n)。例如,正整数6有如下11种不同的划分,所以p(6)=11。6;5+1;4+2;4+1+1;3+3;3+2+1;3+1+1+1;2+2+2_n的划分种数,其中划分大于等于2非递归算法