Spark job server使用调研

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将会终止，其占用资源将被释放。