pyspark_structured_streaming_socket_word_count.py

1from pyspark.sql import SparkSession
2from pyspark.sql.functions import explode
3from pyspark.sql.functions import split
4
5spark = SparkSession \
6    .builder \
7    .appName("StructuredNetworkWordCount") \
8    .getOrCreate()
9
10# Create DataFrame representing the stream of input lines from connection to localhost:9999
11lines = spark \
12    .readStream \
13    .format("socket") \
14    .option("host", "localhost") \
15    .option("port", 9999) \
16    .load()
17
18# Split the lines into words
19words = lines.select(
20   explode(
21       split(lines.value, " ")
22   ).alias("word")
23)
24
25# Generate running word count
26wordCounts = words.groupBy("word").count()
27
28 # Start running the query that prints the running counts to the console
29query = wordCounts \
30    .writeStream \
31    .outputMode("complete") \
32    .format("console") \
33    .start()
34
35query.awaitTermination()