Spark家伙,我对Spark很新,这就是为什么我希望你的帮助确实如此.
我正试图从我的笔记本电脑上安排火花集群上相当简单的工作.尽管它有效,但当我提交它时./spark-submit,它会抛出异常,当我尝试以编程方式进行时.
环境: - Spark - 1个主节点和2个工作节点(独立模式).Spark没有编译,但下载了二进制文件.Spark版本 - 1.0.2 - java版本"1.7.0_45" - 应用程序jar位于任何地方(在客户端和工作节点上的同一位置); - README.md文件也被复制到每个节点;
我正在尝试运行的应用程序:
val logFile = "/user/vagrant/README.md"
val conf = new SparkConf()
conf.setMaster("spark://192.168.33.50:7077")
conf.setAppName("Simple App")
conf.setJars(List("file:///user/vagrant/spark-1.0.2-bin-hadoop1/bin/hello-apache-spark_2.10-1.0.0-SNAPSHOT.jar"))
conf.setSparkHome("/user/vagrant/spark-1.0.2-bin-hadoop1")
val sc = new SparkContext(conf)
val logData = sc.textFile(logFile, 2).cache()
...
所以问题是,当我这样做时,这个应用程序在集群上成功运行:
./spark-submit --class com.paycasso.SimpleApp --master spark://192.168.33.50:7077 --deploy-mode client file:///home/vagrant/spark-1.0.2-bin-hadoop1/bin/hello-apache-spark_2.10-1.0.0-SNAPSHOT.jar
但是当我尝试通过调用以编程方式执行相同操作时,它不起作用 sbt run
这是堆栈跟踪,我在主节点上得到:
14/09/04 15:09:44 ERROR Remoting: org.apache.spark.deploy.ApplicationDescription; local class incompatible: stream classdesc serialVersionUID = -6451051318873184044, local class serialVersionUID = 583745679236071411
java.io.InvalidClassException: org.apache.spark.deploy.ApplicationDescription; local class incompatible: stream …如果,当我遍历RDD时,我需要通过调用外部(阻塞)服务来计算数据集中的值？您认为如何实现这一目标？
val值: Future[RDD[Double]] = Future sequence tasks
我试图创建一个Futures列表,但由于RDD id不是Traversable,Future.sequence不适合.
我只是想知道,如果有人有这样的问题,你是如何解决的？我试图做到的,是得到一个工作节点上的并行性,所以我可以调用外部服务3000每次第二.
可能还有另一种解决方案,更适合火花,就像在单个主机上有多个工作节点一样.
有趣的是,你如何应对这样的挑战？谢谢.
真的需要你的帮助来理解,我做错了什么.
我的实验的目的是以编程方式运行spark作业,而不是使用./spark-shell或./spark-submit(这些都适用于我)
环境:我使用./spark-ec2脚本创建了一个包含1个master和1个worker的Spark Cluster
但是,当我尝试运行打包在jar中的代码时,群集看起来很好:
val logFile = "file:///root/spark/bin/README.md"
val conf = new SparkConf()
conf.setAppName("Simple App")
conf.setJars(List("file:///root/spark/bin/hello-apache-spark_2.10-1.0.0-SNAPSHOT.jar"))
conf.setMaster("spark://ec2-54-89-51-36.compute-1.amazonaws.com:7077")
val sc = new SparkContext(conf)
val logData = sc.textFile(logFile, 2).cache()
val numAs = logData.filter(_.contains("a")).count()
val numBs = logData.filter(_.contains("b")).count()
println(s"1. Lines with a: $numAs, Lines with b: $numBs")
我得到一个例外:
*[info] Running com.paycasso.SimpleApp
14/09/05 14:50:29 INFO SecurityManager: Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
14/09/05 14:50:29 INFO SecurityManager: Changing view acls to: root
14/09/05 14:50:29 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with …