返回 登录
0

Spark的那些外部框架

阅读4187

引言: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来完成这项工作。
            图片描述
  在Spark package站点上注册了Github账号后,可以从“name”下拉菜单中选择你的代码库。
             图片描述
  上面的简短描述和主页最好与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。
              图片描述
  虽然这个界面很简单,但它为管理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:大数据集群计算的生产实践》,点此链接可在博文视点官网查看此书。
                      图片描述

评论