<jupyter_start><jupyter_text>Chapter 8: Structured Streaming<NAME>    May, 2022   Python examples of chapter 8 (page 207 ff) in the book *Learning Spark*<jupyter_code># Import required python spark libraries
import pyspark
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.streaming import StreamingContext

# create a SparkSession
# This requires access to the internet. If executed offline, an error is thrown

spark = (SparkSession \
         .builder \
         .appName("Chapter_8") \
         .getOrCreate())<jupyter_output><empty_output><jupyter_text>Create a streamRun the command `nc -lk 9999`in a terminal window.    All text you type into the terminal will be send as a data stream to port 9999 whenever you hit `Return`.    `nc` stands for *Netcat* and is a simple computer network utility available under Linux, macOS and Windows.   Example of Reading a stream of dataCreating a DataFrame from a text data stream to be received over a socket connection on localhost. Doing a continuous word count on the streaming data and print the results to the console.Whenever text is typed into the `nc` command in the terminal, the text is processed and the word count is printed out in the console of the Jupyter notebook until the spark command `streamingQuery.stop()` is executed.<jupyter_code>lines = (spark
         .readStream.format("socket")
         .option("host", "localhost")
         .option("port", "9999")
         .load()
)

words  = lines.select(split(col("value"), "\\s").alias("word"))
counts = words.groupBy("word").count()

# The directory is created if it does not exist
checkpointDir = "/tmp/checkpoints"

streamingQuery = (counts
                 .writeStream
                 .format("console")
                 .outputMode("complete")  
                 .trigger(processingTime="2 second")
                 .option("checkpointLocation", checkpointDir)
                 .start()
                 )

# Stop the Streaming Query:
streamingQuery.stop()<jupyter_output><empty_output><jupyter_text>Monitoring an Active StreamPage 223 ff.<jupyter_code># Show last progress of the stream. Only shows results when the stream is active
streamingQuery.lastProgress
# Show the actual status of the stream:
streamingQuery.status<jupyter_output><empty_output><jupyter_text>Other Examples Example from Sparks DocumentationSee: https://spark.apache.org/docs/latest/streaming-programming-guide.html     and:https://github.com/apache/spark/blob/v3.2.1/examples/src/main/python/streaming/network_wordcount.pyThe Python code is in the file `network_wordcount.py`.1. Run the program `nc -lk 9999` in a terminal.       This program sends all text entered in the terminal out via port 99992. Run the command `spark-submit network_wordcount.py localhost 9999` in another terminal.3. Each time words are typed in the first terminal, the words are counted in the second terminal4. Terminate both commands with `Cntr-C`.  Example from the Blog *Apache Spark Structured Streaming with Pyspark* from <NAME>**Windowed Streaming:**See: [Link to the blog on Medium](https://medium.com/analytics-vidhya/apache-spark-structured-streaming-with-pyspark-b4a054a7947d)The scenario is that in a retail store sold articles are entered into the terminal whenever they are sold. The article names are transported as a stream to Spark and a Spark query groups the input data by article and counts them and writes the result to the console.<jupyter_code>rawdata = (spark
           .readStream
           .format("socket")
           .option("host", "localhost")
           .option("port", "9999")
           .option("includeTimestamp", True)
           .load()
          )

query = (rawdata
         .select((rawdata.value).alias("product"),(rawdata.timestamp).alias("time"))
         .groupBy(window("time", "1 minutes"), "product")
         .count().sort(desc("window"))
        )
result = (query.writeStream
          .outputMode("complete")
          .format("console")
          .start(truncate=False)
         )

# Stop the Streaming Query:
result.stop()

# Show the actual status of the stream:
result.status<jupyter_output><empty_output><jupyter_text>Streaming Data Sources and SinksPage 226 ff.<jupyter_code># Create dir /tmp/sparkInputDir

inputDirectoryOfFiles = "/tmp/sparkInputDir"

fileSchema = (StructType()
             .add(StructField("key", IntegerType()))
             .add(StructField("value", StringType()))
             )

inputDF =(spark
          .readStream
          .format("csv")
          .schema(fileSchema)
          .option("maxFilesPerTrigger", 1)
          .load(inputDirectoryOfFiles)
         )

inputDF.isStreaming
inputDF.printSchema()
streamingQuery = (inputDF
                 .writeStream
                 .format("console")
                 .outputMode("append")
                 .start()
                 )<jupyter_output><empty_output><jupyter_text>Now copy the files `streamingData_xx.csv` one by one from the directory `data/streamingData` to the direcory `/tmp/sparkInputDir`. Everytime a file is copied, the content is read by the `readStream` and written to the console by the `writeStream`.<jupyter_code># Stop streaming
streamingQuery.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09486543233441497, 'fraction_numerical': 0.011834319526627219, 'mean_word_length': 3.3092105263157894, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '240356', 'license': 'Apache-2.0', 'n_tokens_mistral': 1501, 'n_tokens_neox': 1416, 'n_words': 488, 'path': 'chapter_08_StructuredStreaming.ipynb', 'repo_name': 'ciwin/Spark', 'size': '5261'}