Job Server概述
Spark-jobserver 提供了一个 RESTful 接口来提交和管理 spark 的 jobs、jars 和 job contexts。此项目包含了完整的 Spark job server 的项目,包括单元测试和项目部署脚本。
“Spark as Service”:针对 job 和 contexts 的各个方面提供了 REST 风格的 api 接口进行管理
- 支持 SparkSQL、Hive、Streaming Contexts/jobs 以及定制 job contexts
- 通过集成 Apache Shiro 来支持 LDAP 权限验证
- 支持亚秒级别低延迟的任务通过长期运行的 job contexts
- 可以通过结束 context 来停止运行的作业(job)
- 分割 jar 上传步骤以提高 job 的启动
- 异步和同步的 job API,其中同步 API 对低延时作业非常有效
- 支持 Standalone Spark 和 Mesos、yarn
- Job 和 jar 信息通过一个可插拔的 DAO 接口来持久化
- 命名 RDD 以缓存,并可以通过该名称获取 RDD。这样可以提高作业间 RDD 的共享和重用
- 支持 Scala 2.10 版本和 2.11 版本
git地址: git@github.com:spark-jobserver/spark-jobserver.git
编译、打包、部署
从github中clone此项目的代码,此处选择jobserver-0.6.2-spark-1.6.1版本分支。
首先根据目标平台来创建相应的配置文件:
$JOBSERVER_HOME/config目录下已存在一些配置模板,可以复用这些模板并对其中的配置项做相应的调整。
在conf目录下创建datacloud.conf及datacloud.sh文件,修改其中的配置项。
$JOBSERVER_HOME/bin/server_package datacloud //编译打包
打包成功后,拷贝出job-server.tar.gz到目标运行平台,应该部署安装spark的服务器环境中。
启动、停止服务
$JOBSERVER_HOME/bin/server_start 启动服务,默认监听端口为8090,可在启动前修改datacloud.conf进行配置。
$JOBSERVER_HOME/bin/server_stop停止服务,注意服务停止后,常驻context将停止运行,因此,重启jobserver需要重新创建常驻context。
定制job Project
添加依赖:
resolvers += "Job Server Bintray" at "https://dl.bintray.com/spark-jobserver/maven"
libraryDependencies += "spark.jobserver" %% "job-server-api" % "0.6.2" % "provided"
libraryDependencies += "spark.jobserver" %% "job-server-extras" % "0.6.2" % "provided"```
通过job server来提交的job,必须实现SparkJob相关的接口,这是jobserver复用context机制的前提:
object SampleJob extends SparkJob {
override def runJob(sc:SparkContext, jobConfig: Config): Any = ???
override def validate(sc:SparkContext, config: Config): SparkJobValidation = ???
}“`
runJob定义一个job的具体实现逻辑。
validate在job执行之前做参数进行验证,验证通过后才会调用runjob方法。
Context管理
GET /contexts – 列出当前所有context
POST /contexts/<name> – 创建context
DELETE /contexts/<name> – 停止一个context,并且停止所有在其运行的作业。
PUT /contexts?reset=reboot – 删除所有context,并且从配置中重新重建context。
POST创建context时,可以对context资源进行配置:
dependent-jar-uris=file:///path/a.jar,file:///path/b.jar 此参数用以对依赖jar包进行配置
num-cpu-cores=10 配置cpu资源
memory-per-node=512m 配置内存资源
上述配置会覆盖datacloud.conf相同的配置项,可配置项可参考job-server/src/main/resources/application.conf
Job管理
GET /jobs – 列出所有job,以及job执行状态,包括等待中、运行中、已完成
POST /jobs – 启动新作业,参数?sync=true标志同步等待作业计算结果
GET /jobs/<jobId> – 获取指定job的执行状态和结果
DELETE /jobs/<jobId> – kill特定job
GET /jobs/<jobId>/config – 获取job相关的配置信息
Job Server使用
临时异步方式提交作业
此方式会临时创建context,作业执行完成后,context被删除。
curl -d "input.string = a b c a b see" '172.16.31.63:8092/jobs?appName=test&classPath=spark.jobserver.WordCountExample
{
"status": "STARTED",
"result": {
"jobId": "5453779a-f004-45fc-a11d-a39dae0f9bf4",
"context": "b7ea0eb5-spark.jobserver.WordCountExample"
}
}```
因为是异步方式提交作业,需要主动检查作业的执行结果:
curl localhost:8090/jobs/5453779a-f004-45fc-a11d-a39dae0f9bf4
{
“duration”: “6.341 secs”,
“classPath”: “spark.jobserver.WordCountExample”,
“startTime”: “2015-10-16T03:17:03.127Z”,
“context”: “b7ea0eb5-spark.jobserver.WordCountExample”,
“result”: {
“a”: 2,
“b”: 2,
“c”: 1,
“see”: 1
},
“status”: “FINISHED”,
“jobId”: “5453779a-f004-45fc-a11d-a39dae0f9bf4”
}“`
临时同步方式提交作业
添加sync参数,值为true
curl -d "input.string = a b c a b see" '172.16.31.63:8092/jobs?appName=test&classPath=spark.jobserver.WordCountExample&sync=true'```
请求的response中将包含计算结果数据
{
“result”: {
“b”: 2,
“a”: 2,
“see”: 1,
“c”: 1
}
}“`
常驻context同步方式提交作业
首先创建一个常驻context
curl -d "" '172.16.31.63:8092/contexts/test-context?num-cpu-cores=4&memory-per-node=512m'
基于此context来创建一个job
curl -d "input.string = a b c a b see" '172.16.31.63:8092/jobs?appName=test&classPath=spark.jobserver.WordCountExample&context=test-context&sync=true'
常驻Context集群已为其分配资源,并且会一直处于运行状态,提交作业时,将会复用此Context进行计算。
如果Jobserver服务进行重启,之前的Context将会终止,其占用资源将被释放。