使用PySpark维护从侦听TCP套接字的数据服务器接收的文本数据的运行字数。不接收输出[Windows]

Question

实际上，我希望维护从监听TCP套接字的数据服务器接收的文本数据的运行字数。我正在使用netcat监听套接字，并在木星笔记本中运行python来使用Spark并计算我的字数。我将包括一些链接，可以提供更多关于这个项目的信息，在文章的底部。

我对火花和网猫完全陌生，所以如果我的术语有误导性的话，我会提前道歉。我正在运行windows，因此我使用nc -l -p 9999在命令终端中启动netcat会话。之后，我在笔记本中运行以下代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split


spark = SparkSession \
    .builder \
    .appName("Assignment 7.1") \
    .getOrCreate()

lines = spark \
    .readStream \
    .format("socket") \
    .option("host", "localhost") \
    .option("port", 9999) \
    .load()

# Split the lines into words
words = lines.select(
   explode(
       split(lines.value, " ")
   ).alias("word")
)

# Generate running word count
wordCounts = words.groupBy("word").count()

try:
    query = wordCounts \
        .writeStream \
        .outputMode("complete") \
        .format("console") \
        .start()

    query.awaitTermination()
except KeyboardInterrupt:
    print('Stopping query')

运行此操作后，代码块将永不停止处理。更改的是，我在其中启动netcat会话的终端开始接受文本。然而，输入文本似乎不会产生任何效果。这是一张终端会话的照片。

我尝试通过点击CTRL+C来结束netcat会话，希望它能终止运行在我的朱庇特笔记本中的查询，并给我一个单词计数的输出，但这是行不通的。结束netcat会话就是这样做的，笔记本中的代码块继续处理。

我在想怎样才能让我的字数增加。我真的很感谢你的帮助。

下面是一个指向包含笔记本的GitHub的链接

这里有一个指向这个项目的教程的链接，作为火花文档提供。

这是一个2分30秒的YouTube视频，展示并解释了这个问题。

Answer 1

如果您想处理来自TCP的流数据，我建议您尝试SparkStreaming。这里有一个单词计数的例子

import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext

master = "local[*]"
appName = "Streaming"
sc = SparkContext(master, appName)
ssc = StreamingContext(sc, 1)

#Dstream with real-time data
lines = ssc.socketTextStream("localhost", 9999)
word_counts = lines.flatMap(lambda line: line.split(" ")) \
    .map(lambda word: (word, 1)) \
    .reduceByKey(lambda a, b: a + b)
word_counts.pprint()

ssc.start()
ssc.awaitTermination()