小编说:Spark社区提供了大量的框架和库。其规模及数量都还在不断增加。本文我们将介绍不包含在Spark 核心源代码库的各种外部框架。Spark试图解决的问题涵盖的面很广,跨越了很多不同领域,使用这些框架能帮助降低初始开发成本,充分利用开发人员已有的知识。
- Spark Package
要使用Spark库,你首先必须了解的东西是Spark package。它有点像Spark的包管理器。当你给Spark集群提交job时,你可以到存放Spark package的网站下载任何package。所有package都存放在这个站点。
http://spark-packages.org/
当你想用一个Spark package时,可以在spark-submit命令或者spark- shell命令中增加包选项:
代码语言:javascript复制$ $Spark_HOME/bin/Spark-shell
-packages com.databricks:Spark-avro_2.10:2.0.1
如果使用了--packages选项,Spark package就会自动把它的JAR包添加到你指定的路径下。你不仅能在Spark集群上使用社区的库,还能到公开发布自己的库。如果要把一个Spark package发布到这个托管服务下,必须遵守下列规则:
- 源代码必须放在Github上。
- 代码库的名字必须与包名相同。
- 代码库的主分支必须有README.md文件,在根目录下必须有LICENSE文件。
换句话说,你不需要编译自己的package。即使你用Spark Packages的模板,编译、发布以及版本更新都将由这项服务完成。sbt插件sbt-spark-package(https://github.com/databricks/sbt-spark-packages)对于生成package也非常有用。如果要在你的项目中包含此插件,请务必在sbt项目的project/plugins.sbt文件中写入下面的代码:
代码语言:javascript复制resolvers = "bintray-Spark-packages" at "https://dl.bintray.com/
Spark-packages/maven/"
addSbtPlugin("org.Spark-packages" % "sbt-Spark-packages" % "0.2.3")
发布Spark包时必须提供如下信息,应该把它们写到build.sbt中:
- spName——package的名称。
- sparkVersion——package所依赖的Spark版本。
- sparkComponents——package所依赖的Spark组件列表,例如SQL、MLlib。
- spShortDescription——package的一句话描述。
- spDescription——关于package的完整描述。
- spHomePage——用于描述package的Web页面的URL。
上述6项是你在发布package之前需要提供的信息。一定要发布到package的代码库的主分支上。你可以使用Spark package的托管站点(https://spark- packages.org/)的Web UI来完成这项工作。
package注册网站
在Spark package站点上注册了Github账号后,可以从“name”下拉菜单中选择你的代码库。
选择package的名称
上面的简短描述和主页最好与build.sbt中的描述和主页URL一致。一旦你提交了package,验证过程就开始了。这个过程通常需要几分钟。当验证完成后,你会收到一封邮件,告诉你验证是否成功。如果成功,就可以用前面描述的--package选项下载你的package了。截至2015年11月,Spark package站点上已经有153个package了。下一节将介绍一些库,它们也是支持Spark package形式的,即它们也以Spark package格式分发。
- XGBoost
XGBoost是一个专用于分布式框架的优化库。这个框架由DMLC(Distributed Machine Learning Community,分布式机器学习社区)开发。顾名思义,在DMLC项目下有许多机器学习库,它们在Hadoop和Spark等已有资源上具有高扩展性。XGBoost是基于Gradient Boosting(梯度提升)算法的。决策树提升算法(Tree Boosting)是一种用于分类的集成学习(ensemble learning)算法,它组合使用了决策树与提升算法,是一种轻量而快速的分类算法。关于树集成及树提升算法在此就不展开讲述了,它们都是简单高效的算法:https://xgboost.readthedocs.org/ en/latest/model.html。
虽然当前XGBoost还不能与Spark集成,但是XGBoost的名气使得Spark社区开发了XGBoost的Spark package:http://Spark-packages.org/package/ rotationsymmetry/Sparkxgboost/。
尽管XGBoost核心开发组不支持这个package,你还是可以使用sparkxgboost包体验一下在Spark上的XGBoost的实现。
- spark-jobserver
提交job的流程需要改进,因为对于非工程师来说,这项工作有点难。你需要理解如何用命令行或者其他UNIX命令去提交Spark job。Spark项目现在是使用CLI来提交job的。spark-jobserver提供了一个RESTful API来管理提交到Spark集群的job。因此,这意味着可以在企业内部环境中将Spark作为一个服务启动。最简单的使用spark-observer的方法就是启动一个为之准备的Docker容器。如果你的笔记本上已经有了Docker环境,你需要做的就是输入下面的命令:
代码语言:javascript复制$ docker run -d -p 8090:8090
velvia/spark-jobserver:0.5.2-SNAPSHOT
执行这条命令,spark-jobserver的Docker镜像将被下载,它会在Docker上启动spark-jobserver作为守护进程。可以通过8090端口查看这个服务器的状态。启动之后,可以看到下图所示的Web UI。
Spark job服务器UI
虽然这个界面很简单,但它为管理job提供了足够的信息。job服务器的后台是带有一个本地executor的Spark集群,它有4个线程运行在这个Docker容器配置下。不过对于你的生产环境而言,这样的配置可能远远不够。现在,假设你的job已经通过REST API发送。在spark-jobserver项目目录下有一个著名的单词计数例子。下载这个例子的源码后,用sbt命令编译。如果你的笔记本上没有sbt,请参照http://www.scala-sbt.org/。
代码语言:javascript复制$ git clone
https://github.com/spark-jobserver/Spark-jobserver.git
$ cd spark-jobserver
$ sbt job-server-tests/package
# You can build test package as a jar format under
# job-server-tests/target/scala-2.10/job-server-
# tests2_2.10-0.6.1-SNAPSHOT.jar, though version number
# might be a little bit different
运行一个job,其过程如下:
(1)上传应用的JAR文件。
(2)选择运行在spark-jobserver上的主类。
提交job时不需要每次都编写Spark应用或者编译它,即使你想与其他人共享。spark-jobserver会对如下对象持久化:
- job状态
- job配置
- JAR
因此,一旦你设置了这些信息,就不需要再次重新上传。重要的是,可以通过spark-jobserver与同事共享你的应用的代码。可以用curl命令进行上传:
代码语言:javascript复制$ curl --data-binary @job-server-test/target/scala-2.10/job-server-
tests_2.10.0.6.1-SNAPSHOT.jar
http://<Your Docker Host IP>:8090/jars/test
$ curl 'http://<Your Docker Host IP>:8090/jars'
{
"tests" : "2015-11-12T02:26:50.069-05:00"
}
如果收到上面信息,就可以上传你的JAR。现在,是时候用输入数据启动你的应用了。
代码语言:javascript复制$ curl -d "input.string = takeshi nobita dora suneo suneo nobita" '
http://<Your Docker Host IP>:8090/jobs?appName=test&classPath=spark.
jobserver.WordCountExample'
$ curl 'http://<Your Docker Host IP>:8090/jobs'
{
"duration": "0.448 secs",
"classPath": "spark.jobserver.WordCountExample",
"startTime": "2015-11-12T03:01:12.362-05:00",
"context": "0a518c58-spark.jobserver.WordCountExample",
"status": "FINISHED",
"jobId": "aed9a387-5319-4d8e-ac3d-0f1ce9d4b1a1"
}
你的job应当成功地完成。得到的结果也能通过REST API下载。
代码语言:javascript复制$ curl http://<Your Docker Host IP>:8090/jobs/aed9a387-5319-4d8e-
ac3d-0f1ce9d4b1a1
{
"status": "OK",
"result": {
"takeshi": 1,
"nobita": 2,
"suneo": 2,
"dora": 1
}
}
这就是一个使用spark-jobserver的进程。虽然这个库仍然在开发中,但由于它是一个开源项目,因此可能很快就会被应用到实际场景。如果你打算在内部使用以处理日常数据,那么spark-jobserver是一个不错的选项。
本文选自《Spark:大数据集群计算的生产实践》