Java - Apache Spark 通信答案

【问题标题】：Java - Apache Spark communicationJava - Apache Spark 通信
【发布时间】：2015-06-05 20:32:14
【问题描述】：

我对 Spark 很陌生，正在寻找一些指导 :-)

Java MVC 应用程序与 Spark 通信的典型方式是什么？为简化起见，假设我想计算某个文件中的单词，该文件的名称是通过 GET 请求提供给我的服务器的。

我最初的方法是打开上下文并在我的 MVC 应用程序中的一个类中实现转换/计算。这意味着在运行时我必须想出一个超级 jar 的 spark-core。问题是：

uber jar 重 80mb
我面临与apache spark: akka version error by build jar with all dependencies 相同的问题 (akka.version)
我可以尝试使用阴影来解决它，但感觉这不是正确的方法。

也许 Maven 中的“提供”范围会对我有所帮助，但我正在使用 ant。

如果我的应用程序（如页面中所建议的那样）已经有一个带有实现的 jar（没有任何 spark 库）并且每次收到请求时都使用 spark-submit。我想它会将结果留在某处。

我是否错过了任何中间路线？

【问题讨论】：

标签： java apache-spark

【解决方案1】：

每次使用 spark-submit 有点重，我建议使用某种长时间运行的 Spark Context。我认为您可能正在寻找的“中间道路”选项是让您的工作使用 IBM Spark Kernel、Zepplin 或 Ooyala 的 Spark Job Server 之类的东西。

【讨论】：

谢谢。我认为我们要让 Spring XD 与 Spark 和其他技术集成。

【解决方案2】：

使用部署在 Spark 顶部的中间件服务是一种很好的做法，该服务管理其上下文、作业失败会引发版本和许多其他需要考虑的事情。

我会推荐 Mist。它实现了 Spark 即服务，并创建了一个统一的 API 层，用于在大数据湖之上构建企业解决方案和服务。

Mist 支持 Scala 和 Python 作业执行。

快速入门如下：

将 Mist 包装器添加到您的 Spark 作业中：
斯卡拉示例：

object SimpleContext extends MistJob {
    override def doStuff(context: SparkContext, parameters: Map[String, Any]): Map[String, Any] = {
        val numbers: List[BigInt] = parameters("digits").asInstanceOf[List[BigInt]]
        val rdd = context.parallelize(numbers)
        Map("result" -> rdd.map(x => x * 2).collect())
    }
}

Python 示例：

import mist
class MyJob:
    def __init__(self, job):
        job.sendResult(self.doStuff(job))
    def doStuff(self, job):
        val = job.parameters.values()
        list = val.head()
        size = list.size()
        pylist = []
        count = 0
        while count < size:
            pylist.append(list.head())
            count = count + 1
            list = list.tail()
        rdd = job.sc.parallelize(pylist)
        result = rdd.map(lambda s: 2 * s).collect()
        return result

if __name__ == "__main__":
    job = MyJob(mist.Job())

运行 Mist 服务：

制造迷雾

git clone https://github.com/hydrospheredata/mist.git
cd mist
./sbt/sbt -DsparkVersion=1.5.2 assembly # change version according to your installed spark

创建配置文件

mist.spark.master = "local[*]"
mist.settings.threadNumber = 16

mist.http.on = true
mist.http.host = "0.0.0.0"
mist.http.port = 2003

mist.mqtt.on = false

mist.recovery.on = false

mist.contextDefaults.timeout = 100 days
mist.contextDefaults.disposable = false

mist.contextDefaults.sparkConf = {
    spark.default.parallelism = 128
    spark.driver.memory = "10g"
    spark.scheduler.mode = "FAIR"
}

运行

spark-submit    --class io.hydrosphere.mist.Mist \
                --driver-java-options "-Dconfig.file=/path/to/application.conf" \ target/scala-2.10/mist-assembly-0.2.0.jar

从终端尝试 curl：

curl --header "Content-Type: application/json" -X POST http://192.168.10.33:2003/jobs --data '{"jarPath":"/vagrant/examples/target/scala-2.10/mist_examples_2.10-0.2.0.jar", "className":"SimpleContext$","parameters":{"digits":[1,2,3,4,5,6,7,8,9,0]}, "external_id":"12345678","name":"foo"}'

【讨论】：