<jupyter_start><jupyter_text>Spark Structured Streaming and Delta TablesSpark provides support for streaming data through *Spark Structured Streaming* and extends this support through *delta tables* that can be targets (*sinks*) or *sources* of streaming data.In this exercise, you'll use Spark to ingest a stream of data from a folder of JSON files that consists of simulated status messages from devices. In a real scenario, the data could come from some other real-time source, such as a Kafka queue or an Azure Event Hub. Create a folder for the incoming stream of data1. Ensure this notebook is attached to your Spark pool (using this **Attach to** drop-down list above).2. Run the cell below to create a folder named **data** to which the simulated device data will be written.    > **Note**: The first cell may take some time to run because the Spark pool must be started.<jupyter_code>from notebookutils import mssparkutils

# Create a folder
inputPath = '/data/'
mssparkutils.fs.mkdirs(inputPath)<jupyter_output><empty_output><jupyter_text>Use Spark Structured Streaming to query a stream of data1. Run the cell below to create a streaming dataframe that reads data from the folder based on a JSON schema that includes the name of the device and its status.<jupyter_code>from pyspark.sql.types import *
from pyspark.sql.functions import *

# Create a stream that reads data from the folder, using a JSON schema
jsonSchema = StructType([
  StructField("device", StringType(), False),
  StructField("status", StringType(), False)
])

fileDF = spark.readStream.schema(jsonSchema).option("maxFilesPerTrigger", 1).json(inputPath)<jupyter_output><empty_output><jupyter_text>2. Wait for the cell above to complete.3. When the streaming dataframe has been created, you can apply a transformation query to aggregate the data and write the results to an output stream. Run the following code to filter the incoming stream for errors in the device data, and count the number of errors per device.<jupyter_code>countDF = fileDF.filter("status == 'error'").groupBy("device").count()
query = countDF.writeStream.format("memory").queryName("counts").outputMode("complete").start()
print('Streaming query started.')<jupyter_output><empty_output><jupyter_text>4. The query output is streamed to an in-memory table. Run the cell below to use SQL to query this table and veiw the number of errors per device.<jupyter_code>%%sql
select * from counts<jupyter_output><empty_output><jupyter_text>5. Note that the query returns no data, because we haven't written any device status data there yet.6. Let's fix that by writing some status event data from a couple of simulated devices.<jupyter_code>device_data = '''{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"ok"}
{"device":"Dev2","status":"error"}
{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"error"}
{"device":"Dev2","status":"ok"}
{"device":"Dev2","status":"error"}
{"device":"Dev1","status":"ok"}'''

mssparkutils.fs.put(inputPath + "data.txt", device_data, True)<jupyter_output><empty_output><jupyter_text>7. Run the SQL query again to see the aggregated error counts (if the query still returns no data, wait a few seconds and try again!) There should be one error for device 1, and two errors for device 2.<jupyter_code>%%sql
select * from counts<jupyter_output><empty_output><jupyter_text>8. Review the results, noting the number of errors. Then run the following code to write more device data.<jupyter_code>more_data = '''{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"ok"}
{"device":"Dev1","status":"error"}
{"device":"Dev2","status":"error"}
{"device":"Dev1","status":"ok"}'''

mssparkutils.fs.put(inputPath + "more-data.txt", more_data, True)<jupyter_output><empty_output><jupyter_text>9. Run the SQL query again (waiting a few seconds if necessary) to see the new status events reflected in the aggregations. There should now be two errors for device 1, and three errors for device 2.<jupyter_code>%%sql
select * from counts<jupyter_output><empty_output><jupyter_text>Create a delta tableAzure Synapse Analytics supports the Linux Foundation *Delta Lake* architecture, which builds on Spark Structured Streaming to add support for transactions, versioning, and other useful capabilities.In particular, you can create *delta tables* as a target (or *sink*) for streaming data, or as a *source* of streaming data for downstream queries.To explore this, we'll write the streaming dataframe based on the **data** folder we created previously to a new delta table, which we'll define using a path to a location in the file system.1. Run the cell below to stream the folder data to a delta table.<jupyter_code>delta_table_path = inputPath + 'deltatable'
stream = fileDF.writeStream.format("delta").option("checkpointLocation", inputPath + 'checkpoint').start(delta_table_path)<jupyter_output><empty_output><jupyter_text>2. Now run the next cell to query the delta table to see the data that has been streamed to it. If at first the query returns no data, wait a few seconds and run the cell again).<jupyter_code>df = spark.read.format("delta").load(delta_table_path)
display(df)<jupyter_output><empty_output><jupyter_text>Delta tables enable you to use a feature named *time travel* to view the data at a previous point in time.4. Run the following query to retrieve the initial micro-batch of data that was streamed from the **devdata.txt** file.<jupyter_code>df = spark.read.format("delta").option("versionAsOf", 0).load(delta_table_path)
display(df)<jupyter_output><empty_output><jupyter_text>5. Now that you've finished exploring Spark Structured Streaming and delta tables, stop the stream of data and clean up the files used in this exercise.<jupyter_code>stream.stop()
query.stop()
print("Stream stopped")
mssparkutils.fs.rm(inputPath, True)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10741859684457872, 'fraction_numerical': 0.0062101376300772075, 'mean_word_length': 6.375, 'pattern_counts': {'":': 32, '<': 49, '<?xml version=': 0, '>': 50, 'https://': 0, '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': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '58709', 'license': 'MIT', 'n_tokens_mistral': 1683, 'n_tokens_neox': 1648, 'n_words': 680, 'path': 'streaming/Spark Structured Streaming and Delta Tables.ipynb', 'repo_name': 'iemejia/DP-900T00A-Azure-Data-Fundamentals', 'size': '5958'}