scala - Spark 流 : StreamingContext doesn't read data files

Question

我是 Spark Streaming 的新手，我正在尝试使用 Spark-shell 开始使用它。
假设我在spark-1.2.0-bin-hadoop2.4的根目录下放置了一个名为“dataTest”的目录。

我想在 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”中的一些文件(并且我还尝试重命名该目录中的一些现有文件)。

但不幸的是我没有得到我想要的(即，我没有得到任何输出，所以 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
, took 0,000021 s
0
15/01/15 19:32:46 INFO JobScheduler: Finished job streaming job 1421346766000 ms
.0 from job set of time 1421346766000 ms
15/01/15 19:32:46 INFO MappedRDD: Removing RDD 137 from persistence list
15/01/15 19:32:46 INFO JobScheduler: Total delay: 0,005 s for time 1421346766000
ms (execution: 0,002 s)
15/01/15 19:32:46 INFO BlockManager: Removing RDD 137
15/01/15 19:32:46 INFO UnionRDD: Removing RDD 78 from persistence list
15/01/15 19:32:46 INFO BlockManager: Removing RDD 78
15/01/15 19:32:46 INFO FileInputDStream: Cleared 1 old files that were older tha
n 1421346706000 ms: 1421346704000 ms
15/01/15 19:32:46 INFO ReceivedBlockTracker: Deleting batches ArrayBuffer()

Answer 1

您是否尝试将文本文件从另一个目录移动到正在监视的目录中？为了使文件流正常工作，您可以原子地将文件放入受监控的目录中，这样一旦文件在列表中可见，Spark 就可以读取文件中的所有数据(如果您正在复制，情况可能并非如此)文件进入目录)。

这在 programming guide 的基本来源小节中有详细记录。