Windows 上 StreamingContext.textFileStream 的正确目录路径格式是什么？

Question

我正在尝试执行 spark 流应用程序来处理文件数据流以执行字数统计。 我正在阅读的目录来自 Windows。如图所示，我使用“Users/Name/Desktop/Stream”之类的本地目录。它不是 HDFS。 我在桌面中创建了一个文件夹作为“流”。 我启动了 Spark Streaming 应用程序，然后将一些文本文件添加到文件夹“Stream”中。但是我的 spark 应用程序无法读取文件。它总是给出空洞的结果。 这是我的代码。

//args(0) = local[2]
object WordCount {
def main(args: Array[String]) {
    val ssc = new StreamingContext(args(0), "word_count",Seconds(5))
    val lines = ssc.textFileStream("Users/name/Desktop/Stream")
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
    wordCounts.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

输出：每 5 秒获取一次空数据

17/05/18 07:35:00 INFO Executor: Running task 0.0 in stage 71.0 (TID 35)
-------------------------------------------
Time: 1495107300000 ms
-------------------------------------------

我也尝试将路径指定为 C:/Users/name/Desktop/Stream - 仍然是同样的问题，应用程序无法读取文件。

如果我给出的目录路径不正确，谁能指导一下？

Answer 1

您的代码没问题，所以 only 问题是使用正确的目录路径。请使用file:// 前缀来表示本地文件系统，它将提供file://C:/Users/name/Desktop/Stream。

请一步一步开始，以确认我们的理解处于同一水平。

当您执行 Spark Streaming 应用程序时，创建的目录与您启动该应用程序的目录相同，例如 Stream。一旦您确认该应用程序可以在本地目录中正常工作，我们将对其进行全局修复，以便从 Windows 上的任何目录中读取（如果仍然需要）。

---

还请确保“移动”文件，因为在受监视目录中创建文件的操作必须是原子操作（部分写入会将文件标记为已处理 - 请参阅 StreamingContext）。

文件必须通过从同一文件系统中的另一个位置“移动”它们来写入受监视的目录。

---

正如您在code 中看到的，directory 路径最终将使用 Hadoop 的 File“包装”，因此问题在于说服它接受您的路径：

if (_path == null) _path = new Path(directory)