Spark的那些外部框架

小编说：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命令中增加包选项：

$ $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文件中写入下面的代码：

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环境，你需要做的就是输入下面的命令：

$ 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/。

$ 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命令进行上传：

$ 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。现在，是时候用输入数据启动你的应用了。

$ 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下载。

$ 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：大数据集群计算的生产实践》