标签: spark-jobserver

我知道Apache Livy是从任何地方与spark交互的休息界面.那么使用Apache Livy而不是spark-jobserver有什么好处呢.使用Livy作为替代方案的spark-jobserver有什么缺点.我在互联网上找不到这个.能否请你帮我澄清一下.

谢谢,

我正在Apache Spark上构建一个RESTful API.提供以下Python脚本spark-submit似乎工作正常:

import cherrypy
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('myApp').getOrCreate()
sc = spark.sparkContext

class doStuff(object):
    @cherrypy.expose
    def compute(self, user_input):
        # do something spark-y with the user input
        return user_output

cherrypy.quickstart(doStuff())

但谷歌搜索我看到像Livy和spark-jobserver这样的东西.我阅读了这些项目的文档和一些教程,但我仍然不完全理解Livy或spark-jobserver优于使用CherryPy或Flask或任何其他Web框架的简单脚本的优点.它是关于可扩展性的吗？上下文管理？我在这里错过了什么？如果我想要的是一个用户不多的简单RESTful API,那么Livy或spark-jobserver值得吗？如果是这样,为什么？

我有一个dockerized Spark实例,并使用SJS从我的Spring启动应用程序中提交spark作业.到目前为止一切正常,但现在按照sjs日志提交作业时出现以下异常.

Uncaught exception while reverting partial writes to file /tmp/spark-f6f7e14c-0d89-40b0-b2d8-262278b619db/blockmgr-1eaf0f4d-8451-4eda-a2e3-ca3acafab871/09/temp_shuffle_c8e20306-d111-49b2-b025-d47ba7cac723
java.nio.channels.ClosedByInterruptException
    at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:202)
    at sun.nio.ch.FileChannelImpl.truncate(FileChannelImpl.java:371)
    at org.apache.spark.storage.DiskBlockObjectWriter.revertPartialWritesAndClose(BlockObjectWriter.scala:191)
    at org.apache.spark.util.collection.ExternalSorter$$anonfun$stop$2.apply(ExternalSorter.scala:807)
    at org.apache.spark.util.collection.ExternalSorter$$anonfun$stop$2.apply(ExternalSorter.scala:806)
    at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
    at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
    at org.apache.spark.util.collection.ExternalSorter.stop(ExternalSorter.scala:806)
    at org.apache.spark.shuffle.sort.SortShuffleWriter.stop(SortShuffleWriter.scala:94)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:76)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
    at org.apache.spark.scheduler.Task.run(Task.scala:70)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

我正在运行Docker for mac 1.13 beta,我在磁盘上有足够的空间.感谢您提供与此相关的任何信息.

我是Spark的新用户。我有一个Web服务，该服务允许用户通过从数据库读取并将结果推回数据库来请求服务器执行复杂的数据分析。我已经将那些分析转移到了各种Spark应用程序中。目前，我使用spark-submit来部署这些应用程序。

但是，我很好奇，当我的Web服务器（用Java编写）收到用户请求时，什么是启动相应Spark应用程序的“最佳实践”方法？Spark的文档似乎是使用“ spark-submit”，但我宁愿不将命令传递给终端以执行此操作。我看到了一个替代方案Spark-JobServer，它提供了一个RESTful接口来完成此操作，但是我的Spark应用程序是用Java或R编写的，似乎无法与Spark-JobServer很好地接口。

是否有另一种最佳实践来从Web服务器（Java）中启动Spark应用程序，并等待状态结果（无论作业成功还是失败）？

任何其他人为实现此目的正在做什么的想法都将非常有帮助！谢谢！

我正在使用我拥有的 log4j 属性的 spark-submit 命令来调用 Spark-submit，如下所示：

/opt/spark-1.6.2-bin-hadoop2.6/bin/spark-submit \
--driver-java-options \
"-Dlog4j.configuration=file:/home/test_api/log4j-driver.properties\ --class Test testing.jar

如何--driver-java-options通过 curl（Apache Spark 的 Hidden REST API）提交作业？

我试过这个：

curl -X POST http://host-ip:6066/v1/submissions/create --header "Content-Type:application/json;charset=UTF-8" --data '{
"action" : "CreateSubmissionRequest",
"appArgs" : [ "" ],
"appResource" : "hdfs://host-ip:9000/test/testing.jar",
"clientSparkVersion" : "1.6.2",
"environmentVariables" : {
"SPARK_ENV_LOADED" : "1"
},
"mainClass" : "Test",
"spark.driver.extraJavaOptions" : "-Dlog4j.configuration=file:/home/test_api/log4j-driver.properties",
"sparkProperties" : {
"spark.jars" : "hdfs://host-ip:9000/test/testing.jar",
"spark.app.name" : "Test",
"spark.eventLog.enabled": "true",
"spark.eventLog.dir": "hdfs://host-ip:9000/test/spark-events",
"spark.submit.deployMode" : "cluster",
"spark.master" : "spark://host-ip:7077"
}
}' …

我们正在尝试找到一种方法来加载 Spark (2.x) ML 训练模型，以便根据请求（通过 REST 接口）我们可以查询它并获取预测，例如http://predictor.com:8080/give /我/预测？a = 1，b = 2，c = 3

有一些开箱即用的库可以将模型加载到 Spark 中（假设它在使用 MLWritable 训练后存储在某个地方），然后将其用于预测，但将其包装在作业中并根据每个请求/调用运行它似乎有些过分SparkContext的初始化。

然而，使用 Spark 的优点是我们可以保存 Pipeline 模型并执行相同的特征转换，而无需在 SparkContext 之外实现它。

经过一番挖掘，我们发现Spark-job-server可以通过允许我们为作业服务器初始化“热”spark-context 来帮助我们解决这个问题，因此，我们可以通过调用预测来服务请求使用 Spark-job-server 的 REST API 在现有上下文中执行作业（并获取结果）。

这是 API 化预测的最佳方法吗？由于特征空间的原因，我们无法预先预测所有组合。

或者，我们考虑使用 Spark Streaming 并将预测保存到消息队列中。这允许我们不使用spark-job-server，但它并没有简化整体流程。有人尝试过类似的方法吗？

我使用sbt job-server-api/packages和sbt job-server-tests/packages进行构建.我可以使用jar -cxf命令扩展/解压缩jar

curl --data-binary /opt/hadoop/spark-jobserver/job-server-api/target/scala-2.10/job-server-api_2.10-0.6.1-SNAPSHOT.jar localhost:8090/jars/test
{
  "status": "ERROR",
  "result": "Jar is not of the right format"
}


 curl --data-binary /opt/hadoop/spark-jobserver/job-server-tests/target/scala-2.10/job-server-tests_2.10-0.6.1-SNAPSHOT.jar localhost:8090/jars/test
{
  "status": "ERROR",
  "result": "Jar is not of the right format"
}

sbt version
[info] Loading project definition from /opt/hadoop/spark-jobserver/project
Missing bintray credentials /root/.bintray/.credentials. Some bintray features depend on this.
Missing bintray credentials /root/.bintray/.credentials. Some bintray features depend on this.
Missing bintray credentials /root/.bintray/.credentials. Some bintray features depend on this.
Missing bintray credentials /root/.bintray/.credentials. Some bintray …

exitTotalDF
  .filter($"accid" === "dc215673-ef22-4d59-0998-455b82000015")
  .groupBy("exiturl")
  .agg(first("accid"), first("segment"), $"exiturl", sum("session"), sum("sessionfirst"), first("date"))
  .orderBy(desc("session"))
  .take(500)

org.apache.spark.sql.AnalysisException: cannot resolve '`session`' given input columns: [first(accid, false), first(date, false),  sum(session), exiturl, sum(sessionfirst), first(segment, false)]

它像sum函数一样无法正确找到列名.

使用Spark 2.1

我正在使用Spark Job Server来运行Spark作业,它运行正常.但是当我尝试执行一项大工作(需要超过40秒)时,我收到此错误:

The server was not able to produce a timely response to your request.

是否需要一些配置才能等待服务器应答？我该怎么办？

谢谢

我希望持久化来自spark作业的RDD,以便所有后续作业都可以使用Spark Job Server.这是我尝试过的:

工作1:

package spark.jobserver

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark._
import org.apache.spark.SparkContext._
import scala.util.Try

object FirstJob extends SparkJob with NamedRddSupport {
  def main(args: Array[String]) {
    val conf = new SparkConf().setMaster("local[4]").setAppName("FirstJob")
    val sc = new SparkContext(conf)
    val config = ConfigFactory.parseString("")
    val results = runJob(sc, config)
    println("Result is " + results)
  }

  override def validate(sc: SparkContext, config: Config): SparkJobValidation = SparkJobValid

  override def runJob(sc: SparkContext, config: Config): Any = {

    // the below variable is to be accessed by other jobs: …