pyspark textFileStreaming在textFile工作时无法检测到txt文件

Question

解释我的问题不同：这个问题与标记的问题不同。首先，输入参数已经是一个目录（这是正确的但标记的问题是错误的）。其次，我在流运行期间将txt文件复制到目录中以模拟新的txt文件到达（因此生成新文件而不是该目录中存在的相同文件）

下面是我的问题

---

我有一个目录和txt文件/tmp/a.txt，文件中的内容是

aaa
bbb

我使用 pyspark 并手动将此文件复制到同一目录中，连续（在流式运行期间文件同时创建）

def count(x):
    if x.isEmpty:
        print("empty")
        return
    print(x.count())

sc = SparkContext()
ssc = StreamingContext(sc, 3)
ssc.textFileStream("/tmp/").foreachRDD(count)

输出显示RDD为空

我怎么用

c = sc.textFile("/tmp/").count()
print(c)

显示c为2（与txt文件内容一致）

为什么流媒体不起作用？

Answer 1

您是要获取添加到/tmp/a.txt 文件的新行还是要获取添加到tmp 目录的新文件？

如果是后者尝试用这个替换你的最后一行

ssc.textFileStream("/tmp/*").foreachRDD(count)

Answer 2

我在 scala 中找到了解决方案（仍然无法在 python 中获取新文件）

首先，sc.textFile 和 sc.textFileStream 采用相同的参数，即目录名称。所以上面的代码是对的。

但是，不同之处在于，如果目录存在，sc.textFile 可以获取文件（并且它必须存在，否则会引发InvalidInputException），但在流模式下sc.textFileStream（本地文件系统） ，它要求该目录不存在并且由流程序创建，否则无法拾取新文件（似乎是一个错误，仅存在于本地文件系统中，根据其他经验在HDFS中似乎运行良好）。

另外，根据其他人的经验，如果你删除目录并重新运行程序，回收站也必须清空。

---

但是，在python中这个问题仍然存在，并且在目录中不存在文件时，scala程序只会打印0但python程序会发出警告

WARN FileInputDStream:87 - Error finding new files 
java.lang.NullPointerException

这是我在python和scala中的代码，写新文件的方式都是一样的，所以我就不在这里贴了

python 代码：

if __name__ == "__main__":
    sc = SparkContext()    
    ssc = StreamingContext(sc, 3)
    ssc.textFileStream(path).foreachRDD(lambda x: print(x.count()))
    ssc.start()
    ssc.awaitTermination()

scala 代码：

def main(args: Array[String]): Unit = {  
  val sc = new SparkContext()
  val ssc = new StreamingContext(sc, Seconds(3))
  ssc.textFileStream(params.inputPath).foreachRDD { x =>
    print(x.count())
  }
  ssc.start()
  ssc.awaitTermination()
}