Structured Streaming

In this video, we will talk about spark structured streaming in Databricks. We will learn what is data stream and how to process streaming data using spark structured streaming. We will also understand how to use data stream reader to perform a stream read from a source and how to use and configure data stream writer to perform a streaming write to sink.

A data stream is any data source that grows over time.

For example, a new data in a data stream might correspond to a new JSON log file landing into a cloud storage. Updates to a database captured in a CDC or Change Data Capture feed. Events queued in a pub/sub messaging feed like Kafka.

To process a data stream,usually We have two approaches: either a traditional approach where we reprocess the entire dataset each time we receive a new update to our data. Or another approach would be to write a custom logic to only capture those files or records that have been added since the last time an update was run. And here we can use the spark structured streaming to achieve this goal.

Spark Structured streaming is a scalable streaming processing engine. It allows us to query an infinite data source where automatically detect new data and process the result incrementally into a data sink. A sink is just a durable file system, such as files or tables. But the question is how to interact and query and infinite data source?

Simply by treating it as a table. In fact, the magic behind Spark Structured Streaming is that it allows user to interact with ever-growing data source as if it were just a static table of records. So a new data in the input data stream is simply treated as a new rows appended to a table. And such a table representing an infinite data source is seen as "unbounded" table. As we said, an input data stream could be a directory of files, a messaging system like Kafka, or simply a Delta table. Delta Lake is well integrated with spark structured streaming.

We can simply use spark.readStream() to query the delta table as a stream source, which allows to process all of the data present in the table as well as any new data that arrive later. This creates a streaming data frame on which we can apply any transformation as if it were just a static data frame. Then to persist the result of a streaming query. We need to write them out to durable storage using dataframe.writeStream method. With the writeStream method, we can configure our output. Here for example, we trigger the streaming processing every 2 minutes to check if there are new arriving records, and we choose to append them to the target table.

So again, after another 2 minutes, we will check if there are new arriving data and we append them to the target table. And all this happened thanks to checkpoints created by Spark to track the progress of our streamingprocessing.

Let us take a closer look on these configurations.

When defining a streaming write, the trigger method specify when the system should process the next set of data. And this is called the Trigger Interval. By default, if we don't provide any trigger interval, the data will be processed every half second or we can specify a fixed interval as we did previously. So the data will be processed in micro batches at our specified interval, for example, every 5 minutes. In addition, we can run our stream in a batch mode to process all available data at once using either trigger once option, or availableNow option. In both cases, the trigger will stop on its own once finished to processing the available data. The only difference is that with the trigger Once, all available data will be processed in a singlebatch, compared to multiple micro batches in availableNow option.

Streaming jobs have also output mode similar to static workloads, either: append mode, which is the defaultmode. In this mode, only new appended rows are incrementally appended to the target table with each batch. While in complete mode,the result table is recalculated each time a write is triggered, so the target table is overwritten with each batch.

Databricks creates checkpoints by storing the current state of our streaming job to cloud storage. Checkpointings allow the streaming engine to track the progress of our streaming processing. An important note here is that checkpoints cannot be shared between several streams. A separate checkpoint location is required for every streaming write to ensure processing guarantees.

Structured streaming provide two guarantees. First in case of failure, the streaming agent can resume from where it left off. Thanks to both the check pointing and also a mechanism called Write-ahead logs,they allow to record the offset range of data being processed during each trigger interval, to track our stream progress. Structured streaming also ensures exactly once data processing because the streaming sink are designed to be idempotent. That is multiple writes of the same data, of course identified by the offset, do not result in duplicates being written to the sink. And of course, the two guarantees here only work if the streaming source is repeatable, like cloud based object storage or pub/sub messaging service.

So, taking all together, repeatable data sources and idempotent sinks allows spark structured streamingto ensure end-to-end exactly once semantics under any failure condition.

Lastly, we need to understand that some operations are not supported by streaming data frames.

Yes, it is true that most operations on a streaming data frame are identical to a static data frame,but there are some exceptions to this. Operations such as sorting and deduplication, are either too complex or logically not possible to dow hen working with streaming data. A full discussion of this exception is out of scope of this course. However, there are advanced streaming methods like windowing and watermarking that can help to do such operations.

Structured Streaming (Hands On)

Here, we will explore the basics of working with Spark Structured Streaming to allow incremental processing of data. We will continue using our bookstore dataset that contains Customers, Orders and Books tables. Let us first copy our dataset.

%run /Repos/[email protected]/Databricks-Certified-Data-Engineer-Associate/Includes/Copy-Datasets

To work with data streaming in SQL, we must first use spark.readStream method PySpark API. This is why we are using here a Python notebook.

spark.readStream method allows us to query a Delta table as a stream source. And from there, we can register a temporary view against this stream source. The temporary view created here is a "streaming" temporary view that allows to apply most transformations in SQL the same way as we would with the static data.

Let us first query this streaming temporary view.

Here, what we see is a streaming result. As we can see, the query is still running, waiting for any new data to be displayed here. Generally speaking, we don't display a streaming result unless a human is actively monitoring the output of a query during development or live dashboarding. Let us click Cancel to stop this active streaming query.

Let us now apply some aggregations on this streaming temporary review. Again, because we are querying a streaming temporary view,this becomes a streaming query that executes infinitely, rather than completing after retrieving a singleset of results. And here we are just displaying an aggregation of input as seen by the stream.

None of these records are being persisted anywhere at this point. For streaming queries like this, we can always explore an interactive dashboard to monitor the streamingperformance. Before continuing, let us cancel this active streaming query.

Remember, when working with streaming data, Some operations are not supported like sorting.

We can use advanced methods like windowing and watermarking to achieve such operations, but it is out of scope for this course.

Now, in order to persist incremental results, we need first to pass our logic back to PySpark DataFrame API. Here, we are creating another temporary view.

%sql
CREATE OR REPLACE TEMP VIEW author_counts_tmp_vw AS (
	SELECT
		author,
		count(book_id) as total_books
	FROM books_streaming_tmp_vw
	GROUP BY author
)

And since we are creating this temporary view from the result of a query against a streaming temporary view,. So this new temporary view is also a "streaming" temporary view. The new streaming temporary view has been created.

(
    spark
        .table("author_counts_tmp_vw")
        .writeStream
        .trigger(processingTime='4 seconds')
        .outputMode("complete")
        .option("checkpointLocation", "dbfs:/mnt/demo/author_counts_checkpoint")
        .table("author_counts")
)

In PySpark DataFrame API,We can use the spark.table() to load data from a streaming temporary view back to a DataFrame. Note that spark always load streaming views as a streaming DataFrames, and static views as a static DataFrames,meaning that: incremental processing must be defined from the very beginning with Read logic to support later an incremental writing. Then, we are using DataFrame.writeStream method to persist the result of a streaming query to a durable storage. This allows us to configure the output with the three settings: The trigger intervals, here every 4 seconds. The output mode, either append or complete. For aggregation streaming queries, we must always use "complete" mode to overwrite the table with the new calculation. Finally, the checkpoint location to help tracking the progress of the streaming processing.

We can think about such a streaming query as an always-on incremental query, and we can always explore its interactive dashboard. From this dashboard, we can see that the data has been processed and we can now query our target table.

Our table has been written to the target table, the author_counts table, and we can see that each author has currently only 1 book. And remember, what we see here is not a "streaming" query! simply because we are querying the table directly. I mean, not as a streaming source through a streaming DataFrame. And if we come back to our streaming query, we see that it is still active. In fact, when we execute a streaming query, the streaming query will continue to update as new datacarrives in the source.

To confirm this, let us add new data to our source table.

%sql
INSERT INTO books
values ("B16", "Hands-On Deep Learning Algorithms with Python", "Sudharsan Ravichandiran", "Computer Science", 25),
("B17", "Neural Network Methods in Natural Language Processing", "Yoav Goldberg", "Computer Science", 30),
("B18", "Understanding digital signal processing", "Richard Lyons", "Computer Science", 35)

The Books Table. Let us run this query and see what will happen in our streaming.

We can see that there is new date arriving. Let us queryo ur target table again to see the updated books counts for each author. Now we see some authors having more than 1 book.

Let us come back to our streaming query, and cancel it to see another scenario. Always remember to cancel any active stream in our notebook, otherwise the stream will be always on and prevents the cluster from auto termination.

For our last scenario, we will add some books for new authors to our source table.

%sql
INSERT INTO books
values ("B19", "Introduction to Modeling and Simulation", "Mark W. Spong", "Computer Science", 25),
("B20", "Robot Modeling and Control", "Mark W. Spong", "Computer Science", 30),
("B21", "Turing's Vision: The Birth of Computer Science", "Chris Bernhardt", "Computer Science", 35)

Three records have been inserted. In this scenario, we modify the trigger method to change our query from an always-on query triggeredevery 4 seconds to a triggered incremental batch. We do this using the availableNow trigger option.

(
    spark.table("author_counts_tmp_vw")
        .writeStream
        .trigger(availableNow=True)
        .outputMode("complete")
        .option("checkpointLocation", "dbfs:/mnt/demo/author_counts_checkpoint")
        .table("author_counts")
        .awaitTermination()
)

With this trigger option, the query will process all new available data and stop on its own after execution. In this case, we can use the awaitTermination method to block the execution of any cell in this notebook until the incremental batch's write has succeeded. Let us now run this query to process the three records we have just added. Done. As we can see, with the availableNow trigger option, the query run in a batch mode. It is executed to process all the available data and then stop on its own. Let us finally query the target table again to see the updated data. Yes, it works. Now we have 18 authors instead of 15.

Incremental Data Ingestion

Here, we will talk about incremental data ingestion from files in databricks. We will talk about two methods. Copy Into command and Auto Loader.

By incremental data ingestion, we mean the ability to load data from a new files that have been encountered since the last ingestion. So each time we run our data pipeline, we don't need to reprocess the files we have processed before. We need to process only the new arriving data files. Databricks provides two mechanisms for incrementally and efficiently processing new data files as they arrive in a storage location, which are Copy Into and Auto Loader.

Copy into is a SQL command that allows user to load data from a file location into a Delta table. It loads data idempotently and incrementally. It means each time we run the command, it will load only the new files from the source location. While the files that have been loaded before are simply skipped. The command is pretty simple.

Copy Into a target table From a specific source location. And we specify the format of the source file to load, for example, CSV or parquet, and any related format options. In addition to any option to control the operation of the Copy Into command.

Here, for example, we are loading from CSV files, having a header and a specific delimiter. And in the Copy Options, we are specifying that the schema can be evolved according to the incoming data.

The second method to load data incrementally from files is Auto Loader, which uses structured streaming in Spark to efficiently process new data files as they arrive in a storage location. We can use Auto Loader to load billions of files into a table. Auto loader can scale to support real time ingestion of millions of files per hour. And since it is based on spark structured streaming, Auto Loader uses checkpointing to track the ingestion process and to store metadata of the discovered files. So Auto Loader ensures that data files are processed exactly once. And in case of failure, Auto Loader can resume from where it left off.

And as expected to work with auto loader, we use the readStream and writeStream methods. Auto loader has a specific format of StreamReader called cloudFiles. And in order to specify the format of the source files, we use simply cloudFiles.format option. And with the load function, we provide the location of the source files. In this location, Auto Loader will detect new files as they arrive and queue them for ingestion.

Once we read the files, we write the data into a target table using the StreamWriter, where we provide the location to store the checkpointing information.

Auto loader can automatically configure the schema of our data. It can detect any update to the fields of the source dataset. However, to avoid this inference costs at every startup of our stream, the inferred schema can be stored to a location to be used later. To provide the location where auto loader can store the schema, use the option cloudfiles.schemaLocation. And this location could be simply the same as the checkpoint location. Now we know everything about Auto Loader. The question is when to use Auto Loader ? and when to use Copy Into command ?

Here is an important things to consider when choosing between Copy Into and Auto Loader. If we are going to ingest file in order of thousands, we can use Copy Into command. However, if we are expecting files in order of millions or more over time, use Auto Loader. In addition, auto loader can split the processing into multiple batches so it is more efficient at scale. And Databricks recommends to use Auto Loader as general best practice when ingesting data from a cloud object storage.

Auto Loader (Hands On)

Here, we will explore incremental data ingestion from files using Auto Loader. We will continue working with our bookstore dataset that contains the three tables, customers, orders and books. Let us start by running the copy dataset script.

%run /Repos/[email protected]/Databricks-Certified-Data-Engineer-Associate/Includes/Copy-Datasets

In this demo, we will ingest the new data from orders received in Parquet Files. Let us explore our data source directory.

files = dbutils.fs.ls(f"{dataset_bookstore}/orders-raw")
display(files)

So currently we have only one parquet file in this directory. We are going to use Auto Loader to read the current file in this directory and detect new files as they arrive to ingest them into a target table.

(spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "parquet")
        .option("cloudFiles.schemaLocation", "dbfs:/mnt/demo/orders_checkpoint")
        .load(f"{dataset_bookstore}/orders-raw")
      .writeStream
        .option("checkpointLocation", "dbfs:/mnt/demo/orders_checkpoint")
        .table("orders_updates")
)

To work with Auto Loader,We use the readStream and writeStream methods from PySpark’s Structured Streaming API. The format here is cloudFiles indicating that this is an Auto Loader stream. And in addition, we provide two options cloudFile.format where we precise that we are reading datafiles of type parquet. And the schemaLocation, a directory in which auto loader can store the information of the inferred schema. Then a load method where we provide the location of our data source files. And we are chaining immediately the writeSteam to write the data into a target table, in our case,orders_updates. And of course, we provided the location for storing the checkpoint information, allowing auto loader to track the ingestion process.

<aside> 💡

</aside>

Let us run this command to begin our auto loader stream.

As we can see, the auto loader is a streaming query since it uses spark structured streaming to load data incrementally. So this query will be continuously active and as soon as the new data arrives in our data source, it will be processed and loaded into our target table. Once the data has been ingested to Delta Lake by auto loader, we can interact with it the same way we would with any table.

The data has been loaded well from our source directory. Let us check how many records we have in our table.