scala - Scala Apache Spark中DStream的输出内容

Question

下面的 Spark 代码似乎没有对文件 example.txt 执行任何操作

val conf = new org.apache.spark.SparkConf()
  .setMaster("local")
  .setAppName("filter")
  .setSparkHome("C:\\spark\\spark-1.2.1-bin-hadoop2.4")
  .set("spark.executor.memory", "2g");

val ssc = new StreamingContext(conf, Seconds(1))
val dataFile: DStream[String] = ssc.textFileStream("C:\\example.txt")

dataFile.print()
ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate

我正在尝试使用 dataFile.print() 打印文件的前 10 个元素

一些生成的输出:

15/03/12 12:23:53 INFO JobScheduler: Started JobScheduler
15/03/12 12:23:54 INFO FileInputDStream: Finding new files took 105 ms
15/03/12 12:23:54 INFO FileInputDStream: New files at time 1426163034000 ms:

15/03/12 12:23:54 INFO JobScheduler: Added jobs for time 1426163034000 ms
15/03/12 12:23:54 INFO JobScheduler: Starting job streaming job 1426163034000 ms.0 from job set of time 1426163034000 ms
-------------------------------------------
Time: 1426163034000 ms
-------------------------------------------

15/03/12 12:23:54 INFO JobScheduler: Finished job streaming job 1426163034000 ms.0 from job set of time 1426163034000 ms
15/03/12 12:23:54 INFO JobScheduler: Total delay: 0.157 s for time 1426163034000 ms (execution: 0.006 s)
15/03/12 12:23:54 INFO FileInputDStream: Cleared 0 old files that were older than 1426162974000 ms: 
15/03/12 12:23:54 INFO ReceivedBlockTracker: Deleting batches ArrayBuffer()
15/03/12 12:23:55 INFO FileInputDStream: Finding new files took 2 ms
15/03/12 12:23:55 INFO FileInputDStream: New files at time 1426163035000 ms:

15/03/12 12:23:55 INFO JobScheduler: Added jobs for time 1426163035000 ms
15/03/12 12:23:55 INFO JobScheduler: Starting job streaming job 1426163035000 ms.0 from job set of time 1426163035000 ms
-------------------------------------------
Time: 1426163035000 ms
-------------------------------------------

15/03/12 12:23:55 INFO JobScheduler: Finished job streaming job 1426163035000 ms.0 from job set of time 1426163035000 ms
15/03/12 12:23:55 INFO JobScheduler: Total delay: 0.011 s for time 1426163035000 ms (execution: 0.001 s)
15/03/12 12:23:55 INFO MappedRDD: Removing RDD 1 from persistence list
15/03/12 12:23:55 INFO BlockManager: Removing RDD 1
15/03/12 12:23:55 INFO FileInputDStream: Cleared 0 old files that were older than 1426162975000 ms: 
15/03/12 12:23:55 INFO ReceivedBlockTracker: Deleting batches ArrayBuffer()
15/03/12 12:23:56 INFO FileInputDStream: Finding new files took 3 ms
15/03/12 12:23:56 INFO FileInputDStream: New files at time 1426163036000 ms:

example.txt是格式:

gdaeicjdcg,194,155,98,107
jhbcfbdigg,73,20,122,172
ahdjfgccgd,28,47,40,178
afeidjjcef,105,164,37,53
afeiccfdeg,29,197,128,85
aegddbbcii,58,126,89,28
fjfdbfaeid,80,89,180,82

如 print文件指出:

/**
* 打印此 DStream 中生成的每个 RDD 的前十个元素。这是一个输出
* 运算符，因此此 DStream 将被注册为输出流并在那里实现。
*/

这是否意味着为此流生成了 0 个 RDD？如果想查看 RDD 的内容，则使用 Apache Spark 将使用 RDD 的收集功能。这些是 Streams 的类似方法吗？那么简而言之如何打印到 Stream 的控制台内容？

更新 :

根据@0x0FFF 注释更新代码。 http://spark.apache.org/docs/1.2.0/streaming-programming-guide.html似乎没有给出从本地文件系统读取的示例。这不像使用 Spark 核心那么常见，那里有从文件读取数据的显式示例？

这是更新的代码:

val conf = new org.apache.spark.SparkConf()
  .setMaster("local[2]")
  .setAppName("filter")
  .setSparkHome("C:\\spark\\spark-1.2.1-bin-hadoop2.4")
  .set("spark.executor.memory", "2g");

val ssc = new StreamingContext(conf, Seconds(1))
val dataFile: DStream[String] = ssc.textFileStream("file:///c:/data/")

dataFile.print()
ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate

但是输出是一样的。当我向 c:\\data 添加新文件时dir(与现有数据文件具有相同的格式)，它们不会被处理。我假设 dataFile.print应该打印前 10 行到控制台？

更新2:

也许这与我在 Windows 环境中运行此代码有关？

Answer 1

您误解了 textFileStream 的用法.以下是 Spark 文档中的描述:

创建一个输入流来监控与 Hadoop 兼容的文件系统中的新文件并将它们作为文本文件读取(使用键作为 LongWritable，值作为文本和输入格式作为 TextInputFormat)。

因此，首先，您应该将目录传递给它，其次，运行接收器的节点应该可以使用该目录，因此最好为此目的使用 HDFS。然后当你放一个 新品 文件放入此目录，它将被函数 print() 处理并为其打印前 10 行

更新:

我的代码:

[alex@sparkdemo tmp]$ pyspark --master local[2]
Python 2.6.6 (r266:84292, Nov 22 2013, 12:16:22) 
[GCC 4.4.7 20120313 (Red Hat 4.4.7-4)] on linux2
Type "help", "copyright", "credits" or "license" for more information.
Spark assembly has been built with Hive, including Datanucleus jars on classpath
s15/03/12 06:37:49 WARN Utils: Your hostname, sparkdemo resolves to a loopback address: 127.0.0.1; using 192.168.208.133 instead (on interface eth0)
15/03/12 06:37:49 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 1.2.0
      /_/

Using Python version 2.6.6 (r266:84292, Nov 22 2013 12:16:22)
SparkContext available as sc.
>>> from pyspark.streaming import StreamingContext
>>> ssc = StreamingContext(sc, 30)
>>> dataFile = ssc.textFileStream('file:///tmp')
>>> dataFile.pprint()
>>> ssc.start()
>>> ssc.awaitTermination()
-------------------------------------------
Time: 2015-03-12 06:40:30
-------------------------------------------

-------------------------------------------
Time: 2015-03-12 06:41:00
-------------------------------------------

-------------------------------------------
Time: 2015-03-12 06:41:30
-------------------------------------------
1 2 3
4 5 6
7 8 9

-------------------------------------------
Time: 2015-03-12 06:42:00
-------------------------------------------