【问题标题】：Python Spark Streaming example with textFileStream does not work. Why?带有 textFileStream 的 Python Spark Streaming 示例不起作用。为什么？
【发布时间】：2015-11-29 06:35:38
【问题描述】：

我使用 spark 1.3.1 和 Python 2.7

这是我第一次使用 Spark Streaming。

我尝试使用火花流从文件中读取数据的代码示例。

这是示例的链接： https://github.com/apache/spark/blob/master/examples/src/main/python/streaming/hdfs_wordcount.py

我的代码如下：

conf = (SparkConf()
     .setMaster("local")
     .setAppName("My app")
     .set("spark.executor.memory", "1g"))
sc = SparkContext(conf = conf)
ssc = StreamingContext(sc, 1)
lines = ssc.textFileStream('../inputs/2.txt')
counts = lines.flatMap(lambda line: line.split(" "))\
          .map(lambda x: (x, 1))\
          .reduceByKey(lambda a, b: a+b)
counts.pprint()
ssc.start()
ssc.awaitTermination()

2.txt文件内容如下：

a1 b1 c1 d1 e1 f1 g1 a2 b2 c2 d2 e2 f2 g2 a3 b3 c3 d3 e3 f3 g3

我希望与文件内容相关的内容会出现在控制台中，但什么都没有。每秒都只有这样的文字：

------------------------------------------ 时间：2015-09-03 15:08:18 ------------------------------------------

和 Spark 的日志。

我做错了什么吗？否则为什么它不起作用？

【问题讨论】：

标签： python apache-spark spark-streaming pyspark

【解决方案1】：

我遇到了类似的问题，但我意识到，一旦我将 Streaming 设置为运行，streamingcontext 就会从新文件中获取数据。仅在流式传输启动后才会摄取新放置在源目录中的数据。

其实 pyspark 文档说得很清楚：

文本文件流（目录）

Create an input stream that monitors a Hadoop-compatible file system for new files and reads them as text files. Files must be wrriten to the monitored directory by “moving” them from another location within the same file system. File names starting with . are ignored.

【讨论】：

这是真的！您只需在智能提交作业后将输入文件复制到其中。因为 Streaming 将只处理新来的流数据！

【解决方案2】：

我找到问题了！

我猜问题出在文件系统行为上。我用的是mac。

如果我只是复制它，我的程序看不到文件。我的程序看到了该文件，但它是空的，当我在此文件夹中创建文件然后输入数据时。

如果我创建文件并将其复制到扫描的目录并在未扫描目录的一段时间内执行此操作，最后我的程序会看到文件和其中的任何内容。

我扫描文件的问题文本中的代码也是如此，但我应该扫描目录。

【讨论】：

【解决方案3】：

Json 数据：

{"timestamp": "1571053218000","t1": "55.23","t2": "10","t3": "ON"}

{“时间戳”：“1571053278000”，“t1”：“63.23”，“t2”：“11”，“t3”：“关闭”}

{"timestamp": "1571053338000","t1": "73.23","t2": "12","t3": "ON"}

{"timestamp": "1571053398000","t1": "83.23","t2": "13","t3": "ON"}

从上面的 json 数据中读取的 Pyspark 代码：

from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.sql.types import IntegerType, LongType, DecimalType,StructType, StructField, StringType
from pyspark.sql import Row
from pyspark.sql.functions import col
import pyspark.sql.functions as F
from pyspark.sql import Window

sc = SparkContext.getOrCreate()
spark = SparkSession(sc)
ssc = StreamingContext(sc, 5)

stream_data = ssc.textFileStream("/filepath/")


def readMyStream(rdd):
  if not rdd.isEmpty():
    df = spark.read.json(rdd)
    print('Started the Process')
    print('Selection of Columns')
    df = df.select('t1','t2','t3','timestamp').where(col("timestamp").isNotNull())
    df.show()


stream_data.foreachRDD( lambda rdd: readMyStream(rdd) )
ssc.start()
ssc.stop()

【讨论】：

【解决方案4】：

如果你使用jupyter notebook来执行这个issue，你需要在批处理层运行程序，然后使用jupyter将文本文件上传到指定的文档中。

【讨论】：