我是Spark Streaming的新手,我正在尝试使用Spark-shell开始使用它.假设我有一个名为"dataTest"的目录放在spark-1.2.0-bin-hadoop2.4的根目录中.
我想在shell中测试的简单代码是(在输入$.\ bin\spark-shell之后):
import org.apache.spark.streaming._
val ssc = new StreamingContext(sc, Seconds(2))
val data = ssc.textFileStream("dataTest")
println("Nb lines is equal to= "+data.count())
data.foreachRDD { (rdd, time) => println(rdd.count()) }
ssc.start()
ssc.awaitTermination()
然后,我复制了"dataTest"目录中的一些文件(我还尝试重命名此目录中的一些现有文件).
但不幸的是我没有得到我想要的东西(即,我没有得到任何outpout,所以看起来ssc.textFileStream不能正常工作),只是一些事情,如:
15/01/15 19:32:46 INFO JobScheduler: Added jobs for time 1421346766000 ms
15/01/15 19:32:46 INFO JobScheduler: Starting job streaming job 1421346766000 ms
.0 from job set of time 1421346766000 ms
15/01/15 19:32:46 INFO SparkContext: Starting job: foreachRDD at <console>:20
15/01/15 19:32:46 INFO DAGScheduler: Job 69 finished: foreachRDD at <console>:20 …我从git复制了一个Play框架Scala项目,当我启动命令$ Play时,我收到了这个错误:
Getting org.scala-sbt sbt ${{sbt.version}} ...
:: problems summary ::
:::: WARNINGS
module not found: org.scala-sbt#sbt;${{sbt.version}}
==== local: tried
D:\play-2.1.0\repository\local\org.scala-sbt\sbt\${{sbt.version}}\ivys\ivy.xml
==== Maven2 Local: tried
file://C:\Users\mghesmoune.stage/.m2/repository/org/scala-sbt/sbt/${{sbt.version}}/sbt-${{sbt.version}}.pom
==== typesafe-ivy-releases: tried
http://repo.typesafe.com/typesafe/ivy-releases/org.scala-sbt/sbt/${{sbt.version}}/ivys/ivy.xml
==== Maven Central: tried
http://repo1.maven.org/maven2/org/scala-sbt/sbt/${{sbt.version}}/sbt-${{sbt.version}}.pom
::::::::::::::::::::::::::::::::::::::::::::::
:: UNRESOLVED DEPENDENCIES ::
::::::::::::::::::::::::::::::::::::::::::::::
:: org.scala-sbt#sbt;${{sbt.version}}: not found
::::::::::::::::::::::::::::::::::::::::::::::
:: USE VERBOSE OR DEBUG MESSAGE LEVEL FOR MORE DETAILS
unresolved dependency: org.scala-sbt#sbt;${{sbt.version}}: not found
Error during sbt execution: Error retrieving required libraries
(see D:\play-2.1.0\framework\.\sbt\boot\update.log for complete log)
Error: Could not retrieve sbt ${{sbt.version}} …从Spark的官方文档中可以看出:" Spark确实为两种常见的使用模式提供了两种有限类型的共享变量:广播变量和累加器. "
(http://spark.apache.org/docs/1.2.0/programming-guide.html#shared-variables)
但是," 广播变量允许程序员在每台机器上保留一个只读变量,而不是随副本一起发送它的副本. "
我想要的是一个" 广播读写变量 ",(我的意思是:所有工人和司机都可以访问和修改的共享变量!)有什么办法吗？谢谢
我是Play框架的新手,
我们说Play 2完全是RESTful - 每个连接没有Java EE会话.但是,我们可以通过不同的方式保存数据:会话,Flash或缓存!
不存在任何矛盾吗？!或者我误解了一些事情？!
有人可以向我解释一下吗？
给定一个ArrayBuffer [Double],如何通过维护它们的索引来对其元素进行排序,例如
val arr ArrayBuffer[Double] = ArrayBuffer(4,5.3,5,3,8,9)
结果必须是:
arrSorted = ArrayBuffer(3,4,5,5.3,8,9)
indices = Arraybuffer(3,0,2,1,4,5) //here the data structure doesn't matter, it can be Array, List, Vector, etc.
谢谢
给定一个数组，如下所示：
val stats: Array[(Int, (Double, Int))]=Array((4,(2,5)), (4,(8,4)), (7,(10,0)), (1,(1,3)), (4,(7,9)))
按键分组时，如何获取对的第一个元素的总和！例如，对于键值4，我必须将这些值相加2.0 + 8.0 + 7.0
result = Array((4, 17.0), (7, 10.0), (1, 1.0))
我从这样做开始，但是我不知道如何继续：
stats.groupBy(_._1) mapValues (_.map(_._2)) //....
感谢帮助 ！
我知道在GraphX中我们可以合并两个图形以便例如更新现有网络......但是,作为更新网络的通常操作是将单个节点插入其中.怎么可以在GraphX中进行这样的更新操作？谢谢 !