Step 1: Define input sources

As with batch queries, the first step is to define a DataFrame from a streaming source. However, while reading batch data sources need spark.read to create a DataFrameReader, streaming sources need spark.readStream to create a DataStreamReader. DataStreamReader has most of the same methods as DataFrameReader, therefore you can use it in a very similar way. Here is an example of creating a DataFrame from a text data stream to be received over a socket connection.

# In Python
          spark = SparkSession…
           lines = ( spark
           .readStream.format("socket")
           .option("host", "localhost")
           .option("port", 9999)
           .load() )
          // In Scala 
          val spark = SparkSession...
          val lines = spark
          .readStream.format("socket")
          .option("host", "localhost")
           .option("port", 9999)
           .load())

This code generates the lines DataFrame as an “unbounded table” of newline separated text data read from “localhost:9999”. Note that, similar to batch sources with spark.read, this does not immediately start reading the streaming data; it only sets up the configurations necessary for reading the data once the streaming query is explicitly started.

Besides sockets, Apache Spark natively supports reading data streams from Apache Kafka and all the various file-based formats that DataFrameReader supports (e.g., Parquet, ORC, JSON, etc.). The details of these sources and their supported options are discussed later in this chapter. Furthermore, a streaming query can define multiple input sources, both streaming and batch sources, which can be combined together using DataFrame operations like unions and joins (also discussed later in this chapter).

Step 2: Transform data

Now we can apply the usual DataFrame operations such as splitting the lines into individual words and then count them as shown in the code below:

# In Python
           words = lines.select(split(col("value"), "s").alias("word"))
          counts = words.groupBy("word").count()
          // In Scala
          val words = lines.select(split(col("value"), "s").as("word"))
          val counts = words.groupBy("word").count()

counts is a streaming DataFrame (that is, a DataFrame on unbounded, streaming data) that represents the running word counts that will be computed once the streaming query is started and the streaming input data is continuously processed.

Note that the above DataFrame operations to transform the lines streaming DataFrame would work in the exact same way if lines were a batch DataFrame. In general, most DataFrame operations that can be applied on a batch DataFrame can also be applied on a streaming DataFrame. To understand which operations are supported in Structured Streaming, you have to recognize the two broad classes of data transformations.

Stateless transformations: Operations like select, filter, map, etc. do not require any information from previous rows to process the next row; each row can be processed by itself. The lack of previous “state” in these operation make them stateless. Stateless operations can be applied to both batch and streaming DataFrames. For instance, the flatMap operation in our code snippet is a stateless operation.

Stateful transformations: In contrast, an aggregation operation like count in the above snippet requires maintaining state to combine data across multiple rows. More specifically, any DataFrame operation involving grouping, joining or aggregations are stateful transformations. While many of these operations are supported in Structured Streaming, a few combinations of them are not supported because it is either computationally hard, or infeasible to compute them in an incremental manner.

The stateful operations supported by Structured Streaming and how to manage their state at runtime are discussed later in the chapter.

Step 3: Define output sink and output mode

After transforming the data, we can define how to write the processed output data with DataFrame.writeStream (instead of DataFrame.write for batch data). This creates a DataStreamWriter which, similar to DataFrameWriter, has additional methods to specify the following

Output writing details - where and how to write the output.

Processing details - how to process and how to recover from failures.

Let us first focus on the writing and processing details in the next step. We will focus on the processing details in the next step. For example, to write the final counts to the console, the following snippet shows how:

# In Python
          writer = ( counts.writeStream
          .format("console")
          .outputMode("complete") )
          // In Scala
           val writer = counts.writeStream
          .format("console")
          .outputMode("complete")

Here we have specified “console” as the output streaming sink and “complete” as the output mode. The output mode of a streaming query specifies what part of the updated output to write out after processing new input data. In this example, as a chunk of new input data is processed and the word counts are updated, we can choose to print to the console either the counts of all the words seen till now (that is, complete mode in the above snippet), or only those words that were updates in the last chunk of input data. This is decided by the specified output mode which can have one of the following values (as we already saw in the programming model (see section “The Programming Model of Structured Streaming”):

Append mode: This is the default mode, where only the new rows added to the result table/DataFrame (for example, the counts table) since the last trigger will be outputted to the sink. Semantically, this mode guarantees that any row that is outputted is never going to be changed or updated by the query in the future. Hence, Append mode is supported by only those queries (e.g., stateless queries) that will never modify previously output data. In contrast, our word count query can update previously generated counts, therefore it does not support Append mode.

Complete mode - In this mode, all the rows of the result table/DataFrame will be outputted at the end of every trigger. This is supported by queries where the result table is likely to be much smaller than the input data and is therefore feasible to be retained in memory. For example, our word count supports complete mode because the counts data is likely to be far smaller than that of the input data.

Update mode - In this mode, only the rows of result that were updated since the last trigger will be outputted at the end of every trigger. This is in contrast to the Append mode as the outputted rows may again be modified by the query and outputted again by the future. Most queries support output mode.

More recent details of output modes supported by different queries can be found in the latest Structured Streaming Programming Guide².

Besides writing the output to the console, Structured Streaming natively supports streaming writes to files and Apache Kafka. In addition, you can write to arbitrary locations using the foreachBatch and foreach API methods. In fact, you can use foreachBatch to write streaming outputs using existing batch data sources. The details of these sinks and their supported options are discussed later in this chapter.

Step 4: Specify processing de tails

The final step before starting the query is to specify details of how to process the data. Continuing with our word count example, we are going to specify the processing details as follows.

# In Python
           checkpoint_dir = "..."
           writer2 = ( writer
           .trigger(Trigger.ProcessingTime("1 second"))
           .option("checkpointLocation", checkpoint_dir))
          // In Scala
           val checkpointDir = "..."
          val writer2 = writer
           .trigger(Trigger.ProcessingTime("1 second"))
           .option("checkpointLocation", checkpointDir)

Here we have specified two details using the DataStreamWriter that we had created with DataFrame.writeStream.

Triggering details - When to trigger the discovery and processing of newly available streaming data. There are two options.

ProcessingTime with or without a trigger interval - This is the default triggering mode which processes streaming data in micro-batches. By default, when no trigger is specified, a query triggers the next micro-batch as soon as the previous micro-batch has completed. Alternatively, you can explicitly specify the ProcessTime trigger with an interval, and the query will trigger micro-batches at that fixed internal.

Once - In this mode, the streaming query will execute exactly one micro-batch - it processes all the new data available in a single batch and then stops itself. This is useful when you want to control the triggering and processing from an external scheduler that will restart the query using any custom schedule (e.g., control cost by only executing a query once per day³).

Continuous - This is an experimental mode (as of Spark 3.0) where the streaming query will process data continuously instead of micro-batches. While only a small subset of DataFrame operations allow this mode to be used, it can provide much lower latency (as low as milliseconds) than the micro-batch trigger modes. Refer to the latest Structured Streaming Programming Guide for the most up to date information.

Checkpoint location - This is a directory in any HDFS-compatible file system where a streaming query saves its progress information, that is, what data has been successfully processed. Upon failure, this metadata is used to restart the failed query exactly where it left off. Therefore, setting this option is necessary for failure recovery with exactly-once guarantees.

Step 5: Starting the query

Once everything has been specified, the final step is to start the query, which you can do with the following.

# In Python
          streaming_query = writer2.start()
          // In Scala
          val streamingQuery = writer2.start()

The returned instance of type StreamingQuery represents an active query and can be used to manage the query, which we will cover later in this chapter.

Note that start() is a non-blocking method, so it will return as soon as the query has been started in the background. If you want the main thread to block until the streaming query has terminated, you can use streamingQuery.awaitTermination(). If the query fails in the background with an error, awaitTermination() will also fail with that same exception.

Furthermore, you can wait up to a time out duration using awaitTermination(timeoutMillis). Finally, you can explicitly stop the query with streamingQuery.stop().

Putting it all together

To summarize, here is the complete code for reading streams of text data over a socket, counting the words and printing the counts to the console.

# In Python
           spark = SparkSession...
          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()
           checkpoint_dir = "..."
          streaming_query = ( 
          counts.writeStream
          .format("console")
          .outputMode("complete")
           .trigger(processingTime = "1 second"))
           .option("checkpointLocation", checkpoint_dir)
          .start() )
          streaming_query.awaitTermination()
          // In Scala
          val spark = SparkSession...
          val lines = spark
          .readStream.format("socket")
          .option("host", "localhost")
           .option("port", 9999)
           .load()
          val words = lines.select(split(col("value"), "s").as("word"))
          val counts = words.groupBy("word").count()
           val checkpointDir = "..."
          val streamingQuery = counts.writeStream
          .format("console")
          .outputMode("complete")
           .trigger(Trigger.ProcessingTime("1 second"))
           .option("checkpointLocation", checkpointDir)
          .start()
          streamingQuery.awaitTermination()

After the query has started, a background thread continuously reads new data from the streaming source, processes it, and writes it to the streaming sinks. Next, let’s take a quick peek under the hood at how this is executed.

Querying current status using StreamingQuery

You can query the current health of an active query using the StreamingQuery instance. Here are two methods.

Get current metrics using StreamingQuery.lastProgress() and StreamingQuery.recentProgress() - When a query processes some data in a micro-batch, we consider it to have made some progress. lastProgress() returns information of the last completed micro-batch. For example, printing the returned object (StreamingQueryProgress in Scala/Java or dictionary in Python) duce something like this (in Scala/Java).

{
           "id" : "ce011fdc-8762-4dcb-84eb-a77333e28109",
           "runId" : "88e2ff94-ede0-45a8-b687-6316fbef529a",
           "name" : "MyQuery",
           "timestamp" : "2016-12-14T18:45:24.873Z",
           "numInputRows" : 10,
           "inputRowsPerSecond" : 120.0,
           "processedRowsPerSecond" : 200.0,
           "durationMs" : {
           "triggerExecution" : 3,
           "getOffset" : 2
           },
           "stateOperators" : [ ],
           "sources" : [ {
           "description" : "KafkaSource[Subscribe[topic-0]]",
           "startOffset" : {
           "topic-0" : {
           "2" : 0,
           "1" : 1,
           "0" : 1
           }
           },
           "endOffset" : {
           "topic-0" : {
           "2" : 0,
           "1" : 134,
           "0" : 534
           }
           },
           "numInputRows" : 10,
           "inputRowsPerSecond" : 120.0,
           "processedRowsPerSecond" : 200.0
           } ],
           "sink" : {
           "description" : "MemorySink"
           }
          }

Some of the noteworthy columns are the following

id - Unique identifier tied to a checkpoint location. It stays the same throughout the lifetime of a query (i.e., same across restarts).

runId - Unique identifier for the current (re)started instance of the query. It changes with every restart.

numInputRows - Number of input rows that were processed in the last micro-batch.

inputRowsPerSecond - Current rate at which input rows are being generated at the source (average over the last micro-batch duration).

processedRowsPerSecond - Current rate at which rows are being processed and written out by the sink (average over the last micro-batch duration). If this rate is consistently lower than the input rate, then the query is unable to process data as fast as it is being generated by the source. This is a key indicator of the health of the query.

sources and sink - Provides source/sink-specific details of the data processed in the last batch.

Get current status using StreamingQuery.status() - This provides information on what the background query thread is doing at this moment. For example, printing the returned object will produce something like this (in Scala/Java).

{
           "message" : "Waiting for data to arrive",
           "isDataAvailable" : false,
           "isTriggerActive" : false
          }

Publishing metrics using DropWizard Metrics

Spark supports reporting metrics via a popular library called Dropwizard Metrics⁶. This library allows metrics to be published to many popular monitoring frameworks like Ganglia, Graphite, etc. These metrics are by default not enabled for Structured Streaming queries due to their high volume of reported data. To enable them, apart from configuring the Dropwizard Metrics for Spark (see the Spark Monitoring and Instrumentation Guide⁷ for more details), you also have to explicitly set the SparkSession configuration spark.sql.streaming.metricsEnabled to true before starting the query.

Note that only a subset of the information available through StreamingQuery.lastProgress() is published through DropWizard Metrics. If you want to continuously publish more progress information to arbitrary locations, you have to write custom listeners as discussed next.

Publishing metrics using custom StreamingQueryListeners

StreamingQueryListener is an event listener interface with which you can inject arbitrary logic to continuously publish metrics. This developer API is available only in Scala/Java. There are two steps to using custom listeners.

Define your custom listener - The StreamingQueryListener interface provides three methods that can be defined by your implementation to get three types of events related to a streaming query - start, progress (i.e., a trigger was executed), and termination. Here is an example:

// In Scala
          val myListener = new StreamingQueryListener() {
           override def onQueryStarted(event: QueryStartedEvent): Unit = {
           println("Query started: " + event.id)
           }
           override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {
           println("Query terminated: " + event.id)
           }
           override def onQueryProgress(event: QueryProgressEvent): Unit = {
           println("Query made progress: " + event.progress)
           }
          }
          Add your listener to the SparkSession before starting query
          // In Scala
          spark.streams.addListener(myListener)

After adding the listener, all events of streaming queries running on this SparkSession will start calling the listener’s methods.

Reading from files

Structured Streaming can treat files written into a directory as a data stream. Here is an example.

# In Python

from pyspark.sql.types import *

input_directory_of_json_files = …

file_schema = (StructType()

.add(StructField(“key”, StringType()))

.add(StructField(“value”, StringType())))

input_df = (

spark.readStream

.format(“json”)

.schema(file_schema)

.load(input_directory_of_json_files) )

// In Scala

import org.apache.spark.sql.types._

val inputDirectoryOfJsonFiles = …

val fileSchema = new StructType()

.add(“key”, StringType)

.add(“value”, StringType)

val inputDF = spark.readStream

.format(“json”)

.schema(fileSchema)

.load(inputDirectoryOfJsonFiles)

The returned streaming DataFrame will have the specified schema. Here are a few key points to remember when using files:

All the files must be of the same format and are expected to have the same schema. For example, if the format is json, all the files must be in the JSON format with one JSON record per line. The schema of each the JSON record must be the same as the one specified with readStream. Violation of these assumptions can lead to incorrect parsing (e.g., unexpected null values) or query failures.

Each file must appear in the directory listing atomically, that is, the whole file must be available at once for reading, and once it is available, the file cannot be updated or modified. This is because Structured Streaming will process the file when the engine finds it (using directory listing) and internally mark it as processed. Any changes to that file will not be processed.

When there are multiple new files to process but it can only pick some of them in the next micro-batch (e.g., rate limits), then it will select the files with the earliest file timestamp. Within the micro-batch, however, there is no predefined order of reading the selected files as all of them will be read in parallel.

Note: This streaming file source supports a number of options - (i) file-format-specific options, same as those by supported in batch with spark.read (see Chapter 4, section “Data Sources for DataFrames and SQL Tables”), and (ii) common options supported only with streaming (e.g., maxFilesPerTrigger to limit the files processed rate). See the programming guide⁸ for full details.

Writing to Files

Structured Streaming supports writing streaming query output to files in the same formats as reads. However, it only supports append mode because it is easy to write new files in the output directory (i.e., append data to a directory) but it is hard to modify existing data file (as would be expected of update and complete modes). Similar to batch, it supports partitioning. Here is an example.

# In Python
          output_directory = …
          streaming_query = ( resultDataFrame.writeStream
          .format("parquet")
          .option("path", output_directory)
          .partitionBy("date")
          .start() )
          // In Scala
          val outputDirectory = …
          val streamingQuery = resultDataFrame.writeStream
          .format("parquet")
          .option("path", output_directory)
          .partitionBy("date")
          .start()

Instead of using the path option, you can also specify the output directory directly as start(output_directory).

A few key points to remember:

Structured Streaming achieves end-to-end exactly-once guarantee when writing to files by maintaining a log of the data files that have been written to the directory. This log is maintained in the subdirectory _spark_metadata. Any Spark query on the directory (not its subdirectories) will automatically use the log to read the correct set of data files such that the exactly-once guarantee is maintained (i.e., eliminates duplicate data, or partial files). Note that other processing engines may not be aware of this log and hence may not provide the same guarantee.

If you change the schema of result DataFrame between restarts, then the output directory will have data in multiple schema. These schemas have to be reconciled by the developer when querying the directory.

Reading from Kafka

To perform distributed reads from Kafka, you have to use options to specify information on how to connect to Kafka. Say, you want to subscribe to data from a topic “events”. Here is how you can create a streaming DataFrame.

# In Python
          input_df = (spark
           .readStream
           .format("kafka")
           .option("kafka.bootstrap.servers", 
           "host1:port1,host2:port2")
           .option("subscribe", "events")
           .load())
          // In Scala
          val inputDF = spark
           .readStream
           .format("kafka")
           .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
           .option("subscribe", "events")
           .load()

The returned DataFrame will have the following schema.

Column Name	Column Type	Description
key	binary	Key data of the record as bytes.
value	binary	Value data of the record as bytes.
topic	string	Kafka topic the record was in. This is useful when subscribed to multiple topics.
partition	int	Kafka topic’s partition the record was in.
offset	long	Offset value of the record.
timestamp	long	Timestamp associated with the record.
timestampType	int	Enumeration for the type of the timestamp associated with the record. See Kafka documentation.

You can also choose to subscribe to multiple topics, a pattern of topics or even a specific partition of a topic. Furthermore, you can choose to read only new data in the subscribed topic, or process all the available data in that topic. Lastly, you can even read Kafka data from batch queries, that is, treat Kafka topics like tables. See the online Kafka Integration Guide for more details¹⁰.

Writing to Kafka

For writing to Kafka, Structured Streaming expects the result DataFrame to have a few columns of specific names and types.

Column Name	Column Type	Description
key (optional)	string or binary	If present, the bytes will be written as the Kafka key, otherwise the key will be empty.
value (required)	string or binary
topic (required only if topic not specified as option)	string	If “topic” is not specified as an option, this determines the topic to write the key-value to. This is useful for fanning out the writes to multiple topics. If “topic” option has been specified, this value is ignored.

You can write to Kafka in all three modes, though Complete mode is not recommended as it will repeatedly output the same records. Here is a concrete example of writing the output of our earlier word count query into Kafka in update mode.

# In Python
           counts = … # DataFrame[word: string, count: long]
           streaming_query = (
          counts.selectExpr(
          "cast(word as string) as key", 
          "cast(count as string) as value")
          .writeStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
           .option("topic", "word_counts")
           .outputMode("update")
          .option("checkpointLocation", checkpoint_dir)
          .start() )
          // In Scala
          val counts = … // DataFrame[word: string, count: long]
           val streamingQuery = counts.selectExpr(
          "cast(word as string) as key", 
          "cast(count as string) as value")
          .writeStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
           .option("topic", "word_counts")
           .outputMode("update")
           .option("checkpointLocation", checkpointDir)
          .start()

See the online Kafka Integration Guide for more details¹¹.

Writing to any storage systems

There are two operations that allow you to write the output of a streaming query to arbitrary storage systems - foreachBatch and foreach. They have slightly different use cases: while foreach allows custom write logic on every row, foreachBatch allows arbitrary operations and custom logic on the output of each micro-batch. Let’s understand their usage in more detail.

Using foreachBatch

foreachBatch(...) allows you to specify a function that is executed on the output of every micro-batch of a streaming query. It takes two parameters: a DataFrame or Dataset that has the output of a micro-batch and the unique identifier of the micro-batch. Here is an example. Say we want to write the output of our earlier word-count query to Apache Cassandra¹². As of Spark Cassandra Connector¹³ 2.4.2, there is no support for writing streaming DataFames. But you can use the connector’s batch DataFrame support to write the output of each batch (i.e., updated word counts) to Cassandra, as shown below:

# In Python
          host_addr = "<ip address>"
          keyspace_name = "<keyspace>"
          table_name = "<tableName>"
          spark.conf.set("spark.cassandra.connection.host", host_addr)
           def write_counts_to_cassandra(updated_counts_df, batch_id):
          # Use Cassandra batch data source to write the updated counts
           ( updated_counts_df.write
           .format("org.apache.spark.sql.cassandra")
           .mode("append")
           .options(table=table_name, keyspace=keyspace_name)
           .save() )
           streaming_query = ( 
          counts.writeStream
          .foreachBatch(write_counts_to_cassandra)
          .outputMode("update")
          .option("checkpointLocation", checkpointDir)
          .start() )
          // In Scala
          val hostAddr = "<ip address>"
          val keyspaceName = "<keyspace>"
          val tableName = "<tableName>"
          spark.conf.set("spark.cassandra.connection.host", hostAddr)
          def writeCountsToCassandra(updatedCountsDF: DataFrame, batchId: Long) {
           // Use Cassandra batch data source to write the updated counts 
           updatedCountsDF.write
           .format("org.apache.spark.sql.cassandra")
           .options(Map("table" -> tableName, "keyspace" -> keyspaceName))
           .mode("append")
           .save()
          }
          val streamingQuery = counts.writeStream
           .foreachBatch(writeCountsToCassandra)
           .outputMode("update")
           .option("checkpointLocation", checkpointDir)
           .start()

With foreachBatch, you can do the following:

Reuse existing batch data sources - As shown in the earlier example, with foreachBatch, you can use existing batch data sources (i.e., sources that support writing batch DataFrames) to write the output of streaming queries.

Write to multiple locations - If you want to write the output of a streaming query to multiple locations (e.g., an OLAP data warehouse and an OLTP database), then you can simply write the output DataFrame/Dataset multiple times. However, each attempt to write can cause the output data to be recomputed (including possible re-reading of the input data). To avoid recomputations, you should cache the batchOutputDataFrame, write it to multiple locations, and then unpersist it.

 # In Python
           def write_counts_to_multiple_locations(updated_counts_df, batch_id):
           updatedCountsDF.persist()
           updatedCountsDF.write.format(...).save() # location 1
           updatedCountsDF.write.format(...).save() # location 2
           updatedCountsDF.unpersist()
          // In Scala
           def writeCountsToMultipleLocations(
           updatedCountsDF: DataFrame, 
           batchId: Long) {
           updatedCountsDF.persist()
           updatedCountsDF.write.format(...).save() // location 1
           updatedCountsDF.write.format(...).save() // location 2
           updatedCountsDF.unpersist()
          }

Apply additional DataFrame operations - Many DataFrame API operations are not supported¹⁴ on streaming DataFrames because Structured Streaming does not support generating incremental plans in those cases. Using foreachBatch, you can apply some of these operations on each micro-batch output. However, you will have to reason about the end-to-end semantics of doing that operation yourself.

Note: foreachBatch only provides at-least-once write guarantees. You can get exactly-once guarantee by using the batchId to deduplicate multiple writes from re-executed micro-batches.

Using foreach

If foreachBatch is not an option (for example, corresponding batch data writer does not exist), then you can express your custom writer logic using foreach. Specifically, you can express the data writing logic by dividing it into three methods: open, process, and close. Structured Streaming will use these methods to write each partition of the output records. abstract example.

# In Python
          class Foreach_writer:
          def open(self, partition_id, epoch_id):
           # Open connection to data store
           # Return True if write should continue
           # This method is optional in Python 
           # If not specified, the write will continue automatically
           return True
          def process(self, row):
           # Write string to data store using opened connection
           # This method is NOT optional in Python
           pass
          def close(self, error):
           # Close the connection.
           # This method is optional in Python
           pass
          streaming_df.writeStream.foreach(Foreach_writer()).start()
          // In Scala
          val foreachWriter = new ForeachWriter[String] {
           def open(partitionId: Long, version: Long): Boolean = {
           // Open connection to data store
           // Return true if write should continue
           }
           def process(record: String): Unit = {
           // Write string to data store using opened connection
           }
           def close(errorOrNull: Throwable): Unit = {
           // Close the connection
           }
           }
          streamingDatasetOfString.writeStream
          .foreach(foreachWriter)
           .start()

The detailed semantics of these methods as executed are discussed in the online programming guide¹⁵.

Reading from any storage system

Unfortunately, as of Spark 2.4, the APIs to build custom streaming sources and sinks are not public and stable. Hence, there is no official way to read from arbitrary storage systems. This may change with Spark 3.0 in which the DataSourceV2 initiative is attempting to stabilize those APIs. Please refer to the latest release notes for the most up-to-date information.

Distributed and Fault-tolerant State Management

Recall from Chapter 1 and 2 that a Spark application running in a cluster has a driver and multiple executors. Spark’s scheduler running in the driver breaks down your high-level operations into smaller tasks, puts them in task queues and, as resources become available, the executors pull the tasks from the queues to execute them. Each micro-batch in a streaming query essentially performs one such set of tasks that read new data from streaming sources and writes updated output to streaming sinks. For stateful stream processing queries, besides writing to sinks, each micro-batch of tasks generate intermediate state data which will be consumed by the next micro-batch. This state data generation is completely partitioned and distributed (as all reading, writing and processing in Spark) and it is cached in the executor memory for efficient consumption. This is illustrated in Figure 9-6 which shows how the state is managed in our original streaming word count query.

Figure 7-6. Distributed state management in Structured Streaming

Each micro-batch reads a new set of words, shuffles them within the executors to group them, computes the counts within the micro-batch and finally adds them to the running counts to produce the new counts. This new count is both, the output as well as the state for the next micro-batch, and hence it is cached in the executor. The next micro-batch data is grouped between executors in exactly the same way as before so that each word is always process by the same executor and can therefore locally read and update its running count.

However, it is not sufficient to just keep this state in memory as any failure (either of an executor, or entire application) will cause the in-memory state to be lost. To avoid loss, we synchronously save the key-value state update as change logs in the checkpoint location provided by the user. These changes are co-versioned with the offsets ranges processed in each batch, and the necessary version of the state can be automatically reconstructed back by reading the checkpointed logs. In case of any failure, Structured Streaming is able to re-execute the failed micro-batch by reprocessing the same input data along with the same state that it had before that micro-batch, thus producing the same output data as it would have had if there were no failure. This is critical for ensuring end-to-end exactly once guarantees.

To summarize, for all stateful operations, Structured Streaming ensures the correctness of operation by automatically saving and restoring the state in a distributed manner. Depending on the stateful operation, all you may have to do is tune the state cleanup policy such that old keys and values can be automatically dropped from the cached state. This is what we will discuss next.

Types of Stateful Operations

The essence of streaming state is to retain summarized information of past data. Sometimes old summaries need to be cleaned up from the state to make room for new summaries. Based on how this state is cleaned up, there are two types of stateful operations:

Managed stateful operations - They automatically identify and cleanup old state, based on a operation-specific definition of “old.” You can tune what is defined as old in order to control the resource usages (e.g., executor memory used to store state). The operations that fall in this category are

Streaming aggregations

Stream-stream joins

Streaming deduplication

Unmanaged stateful operations - They let you define your custom state cleanup logic. This category has the following operations.

MapGroupsWithState

FlatMapGroupsWithState

These operations allow you define arbitrary stateful operations (e.g., sessionization, etc.)

Each of these operations are discussed in detail in the following sections.

Semantic Guarantees with Watermarks

Before we conclude this section about watermarks, let’s understand the precise semantic guarantee that watermarking provides. A watermark of 10 minutes guarantees that the engine will never drop any data that is delayed by less than 10 minutes compared to the latest event time seen in the input data. However, the guarantee is strict only in one direction. Data delayed by more than 10 minutes is not guaranteed to be dropped, that is, it may get aggregated. Whether an input record more than 10 minutes late will be actually aggregated or not depends on the exact timing of when the record was received and when the micro-batch processing it was triggered.

Inner joins with optional watermarking

Say we have redefined our impressions DataFrame to be a streaming DataFrame. To get the stream of matching impressions and their corresponding clicks, you will use the same code as we have seen earlier for static joins and stream-static joins.

# In Python
          # Streaming DataFrame [adId: String, impressionTime: Timestamp, …]
          impressions = spark.readStream. … 
          # Streaming DataFrame[adId: String, clickTime: Timestamp, …]
          clicks = spark.readStream. … 
          matched = impressions.join(clicks, "adId")
          // In Scala
          // Streaming DataFrame [adId: String, impressionTime: Timestamp, …] 
          val impressions = spark.readStream. … 
          // Streaming DataFrame[adId: String, clickTime: Timestamp, …] 
          val clicks = spark.readStream. … 
          val matched = impressions.join(clicks, "adId")

Even though the code is the same, the execution is completely different. When this query is executed, the processing engine will recognize it to be a stream-stream join instead of a stream-static join. The engine will buffer all clicks and impressions as state, and will generate a matching impression-and-click as soon as a received click matched a buffered impression (or vice versa, depending on which was received first). Let’s visualize how this inner join works using an example timeline of events.

Figure 7-12. Illustrative timeline of clicks, impressions and their joined output

In Figure 9-12, the blue dots represent the event times of impression and click events that were received over time across different micro-batches (separated by the dashed grey lines). For the purposes of this illustration, assume that each event was actually received at the same wall clock time as the event time. Note the different scenarios under which the related events are being joined. Both events with adId = ⧮ were received in the same micro-batch, so their joined output was generated by that micro-batch. However, for adId = ⧉ impression was received at 12:04, much earlier than its corresponding click at 12:13. Structured Streaming would first receive the impression at 12:04 and buffer it in the state. For each received click, the engine will try to join it with all buffered impressions and vice versa. Eventually, at a later micro-batch running around 12:13, the engine would receive the click for adId = ⧉ and generate the joined output.

However, in this query, we have not given any indication on how long the engine should buffer an event to find a match. Therefore, the engine may buffer an event forever and accumulate an unbounded amount of streaming state. To limit the streaming state maintained by stream-stream joins, you need to know the following information about your use case:

What is the time range between the generation of the two events at their respective sources? In the context of our use case, let’s assume that a click can occur within 0 seconds to 1 hour after the corresponding impression.

What is the maximum duration an event can be delayed in transit between the source and the processing engine? For example, ad clicks from a browser may get delayed due to intermittent connectivity and arrive much later and out-of-order than expected. Let’s say, that impressions and clicks can be delayed by at most 2 and 3 hours, respectively.

These delay limits and event time constraints between matching events can be encoded in the DataFrame operations using watermarks and time range conditions. In other words, you will have to do the following additional steps in the join to ensure state cleanup.

Define watermark delays on both inputs such that the engine knows how delayed the input can be (similar to streaming aggregations).

Define a constraint on event-time across the two inputs such that the engine can figure out when old rows of one input is not going to be required (i.e., will not satisfy the time constraint) for matches with the other input. This constraint can be defined in one of the following two ways:

Time range join conditions (e.g., join condition = “leftTime BETWEEN rightTime AND rightTime + INTERVAL 1 HOUR”)

Join on event-time windows (e.g., join condition = “leftTimeWindow = rightTimeWindow”).

In our use case of advertisements, our inner join code will get a little bit more complicated, as shown below.

# In Python
          # Define watermarks
          impressions_with_watermark = ( impressions
           .selectExpr("adId AS impressionAdId", "impressionTime")
           .withWatermark("impressionTime", "2 hours") )
          clicks_with_watermark = ( clicks
           .selectExpr("adId AS clickAdId", "clickTime")
           .withWatermark("clickTime", "3 hours") )
          # Inner join with time range conditions
          impressions_with_watermark.join(
           clicks_with_watermark,
           expr(""" 
           clickAdId = impressionAdId AND 
           clickTime BETWEEN impressionTime AND impressionTime + interval 1 hour"""))
          // In Scala
          // Define watermarks
          val impressionsWithWatermark = impressions
           .selectExpr("adId AS impressionAdId", "impressionTime")
           .withWatermark("impressionTime", "2 hours ")
          val clicksWithWatermark = clicks
           .selectExpr("adId AS clickAdId", "clickTime")
           .withWatermark("clickTime", "3 hours")
          // Inner join with time range conditions
          impressionsWithWatermark.join(
           clicksWithWatermark,
           expr(""" 
           clickAdId = impressionAdId AND 
           clickTime BETWEEN impressionTime AND impressionTime + interval 1 hour"""))

With these time constraints for each event, the processing engine can automatically calculate how long events need to be buffered for generating correct results and when the events can be dropped from state. For example, it will evaluate the following (also illustrated in Figure 9-x).

Impressions need to be buffered for at most 4 hours (in event-time) as a 3-hour-late click may match with an impression made 4 hours ago (i.e., 3-hour-late + up to 1 hour delay between the impression and click).

Conversely, clicks need to be buffered for at most 2 hours (in event-time) as a 2-hour-late impression may match with click received 2 hours ago.

Figure 7-13. Structured Streaming automatically calculates thresholds for state clean up using watermark delays and time range conditions

There are a few key points to remember about inner joins.

For inner joins, specifying watermarking and event-time constraints are both optional. In other words, at the risk of potentially unbounded state, you may choose not to specify them. Only when both are specified, you will get state cleanup.

Similar to the guarantees provided by watermarking on aggregations, a watermark delay of “2 hours” guarantees that the engine will never drop or not match any data that is less than 2 hours delayed. But data delayed by more than 2 hours may or may not get processed.

Outer joins with Watermarking

The earlier inner join will output only those ads for which both events have been received. In other words, ads that received no click would not be reported at all. Instead you may want all ad impressions to be reported, with or without the associated click data, to enable additional analysis later (e.g., click through rates). This brings us to stream-stream outer joins. All you need to do is specify the outer join type. That is, the following code:

# In Python
          # Left outer join with time range conditions
          impressions_with_watermark.join(
           clicks_with_watermark,
           expr(""" 
           clickAdId = impressionAdId AND 
           clickTime BETWEEN impressionTime AND impressionTime + interval 1 hour"""),
           "leftOuter") # only change: set the outer join type
          // In Scala
          // Left outer join with time range conditions
          impressionsWithWatermark.join(
           clicksWithWatermark,
           expr(""" 
           clickAdId = impressionAdId AND 
           clickTime BETWEEN impressionTime AND impressionTime + interval 1 hour"""),
           "leftOuter") // only change: set the outer join type

As expected of outer joins, this query will start generating output for every impression, with or without (i.e., using NULLS) the click data. However, outer joins have a few additional points to note.

Unlike inner joins, the watermark delays and event-time constraints are not optional for outer joins. This is because for generating the NULL results, the engine must know when an event is not going to match with anything else in the future. Hence, the watermarks and event-time constraints must be specified for generating correct outer join results (along with state cleanup).

Consequently, the outer NULL results will be generated with a delay as the engine has to wait for a while to ensure that there neither were nor would be any matches. This delay is the maximum buffering time (with respect to event-time) calculated by the engine for each event as discussed in the earlier section (i.e., 4 hours for impressions and 2 hours for clicks).

Processing-time Timeouts

Processing-time Timeout is based on the system time (also known as the wall clock time) of the machine running the streaming query and is defined as follows. If a key had last received data at system timestamp T, and the current timestamp is more than (T + timeout duration), then the function will be called again with an empty new data iterator.

Let’s understand how to use timeouts by updating our user example to remove a user’s state based on 1 hour of inactivity. We will make three changes:

In mapGroupsWithState, we will specify the timeout as ProcessingTimeTimeout.

In the state update function, before updating the state with new data, we have to check whether the state has timed out or not. Accordingly, we will update or remove the state.

In addition, every time we update the state with new data, we will set the timeout duration.

// In Scala
          def updateUserStatus(
 userId: String, 
 newActions: Iterator[UserAction],
 state: GroupState[UserStatus]): UserStatus = {
          if (!state.hasTimeoutOut) { // was not called due to timeout
 val userStatus = state.getOption.getOrElse {
           new UserStatus()
          }
          newActions.foreach { action => userStatus.updateWith(action) }
          state.update(userStatus) 
          state.setTimeoutDuration("1 hour") // set processing-time timeout duration
          return userStatus
           } else {
           val userStatus = state.get()
           state.remove() // remove state when timed out
          return userStatus.asInactive() // return inactive user’s status
           }
          }
          val latestStatuses = userActions
          .groupByKey(userAction => userAction.userId) 
          .mapGroupsWithState(
          GroupStateTimeout.ProcessingTimeTimeout)(
          updateUserStatus)

This query will automatically cleanup the state of users for whom the query has not processed any data for more than an hour. However, there are a few points to note about timeouts:

The timeout set by the last call to the function is automatically cancelled when the function is called again, either for the new received data or for the timeout. Hence, whenever the function is called, the timeout duration or timestamp needs to be explicitly set to enable the timeout.

Since the timeouts are processed during the micro-batches, the timing of the timeout is imprecise and depends heavily on the trigger interval and micro-batch processing times. Therefore, it is not advised to use timeouts for precise timing control.

While processing-time timeouts are simple to reason about, they are not robust towards slowdowns and downtimes. If the streaming query suffers a downtime for more than 1 hour, then after restart, all the keys in the state will suffer timeouts as by the system time — for more than one hour has passed since each key received data. Similar wide-scale timeouts can occur if the query processes data slower than they are arriving at the source (e.g., arriving and getting buffered in Kafka). For example, if the timeout is 5 minutes, then a sudden drop in processing rate (or spike in data arrival rate) that causes 5 min lag could produce spurious timeouts. To avoid such issues, we can use event-time timeout, which we will discuss next.

Event-time Tim eout

Instead of the system clock time, event-time timeout is based on the event time in the data (similar to time-based aggregations) and watermark defined on the event time. If a key is configured with a specific timeout timestamp of T (i.e., not a duration), then that key timeouts when the watermark exceeds T and no new data was received for that key since the last time the function was called. Recall that the watermark is a moving threshold that lags behind the maximum event-time seen while processing the data. Hence, unlike system time, the watermark moves forward in time at the same rate as the data is processed. Therefore, unlike processing-time timeouts, any slowdown or downtime in the query processing will not cause spurious timeouts.

Let’s modify our example to use event-time timeout. In addition to the changes we already made for using processing-time timeout, we will make the following changes:

Define watermarks on the input dataset (assume that the class UserAction has an eventTimestamp field). Set the watermark threshold as an acceptable threshold by which input data can be late and out-of-order.

Update mapGroupsWithState to use EventTimeTimeout.

Update the function to set the threshold timestamp at which the timeout will occur. Note that event-time timeout does not allow setting a timeout duration like the processing-time timeout. We will discuss the reason for this later. In this example, we will calculate this timeout as the current watermark + 1 hours.

Here is the updated example.

// In Scala
          def updateUserStatus(
 userId: String, 
 newActions: Iterator[UserAction],
 state: GroupState[UserStatus]):UserStatus = {
          If (!state.hasTimeoutOut) { // was not called due to timeout
 val userStatus = if (state.getOption.getOrElse {
           new UserStatus()
          }
          newActions.foreach { action => userStatus.updateWith(action) }
          state.update(userStatus)
          // set the timeout timestamp to the current watermark + 1 hour
          state.setTimeoutTimestamp(state.getCurrentWatermarkMs, "1 hour") 
          return userStatus
           } else {
           val userStatus = state.get()
           state.remove()
          return userStatus.asInactive()
           }
          }
          val latestStatuses = userActions
           .withWatermark("eventTimestamp", "10 minutes") // define watermark
          .groupByKey(userAction => userAction.userId) 
          // use event-time timeout
          .mapGroupsWithState(
          GroupStateTimeout.EventTimeTimeout)(
          updateUserStatus)

This query will be much more robust to spurious timeouts caused by restarts and processing delays. Here are a few points to note about event-time timeouts:

Unlike the previous example with processing-timeouts, we have used GroupState.setTimeoutTimestamp instead of GroupState.setTimeoutDuration. This is because with processing time-timeouts, the duration is sufficient to calculate the exact future timestamp (i.e., current system time + specified duration) when the timeout would occur. This is not the case for event-time as different applications may want to use different strategies to calculate the threshold timestamp. In this example, we simply calculated the timeout timestamp based on the current watermark. A different application may instead choose to calculate a key’s timeout timestamp based on the maximum event-time timestamp seen for that key (tracked and saved as part of the state).

The timeout timestamp must be set to a value larger than the current watermark. This is because the timeout is expected to happen when the timestamp crosses the watermark, and it’s illogical to set the timestamp to a value already larger than the current watermark.

Before we move on from timeouts, one last thing to remember is that you can use the timeout mechanisms for more creative processing than fixed-duration timeouts. For example, you can implement an approximately periodic task (say, every hour) on the state by saving the last task execution timestamp in the state and using that to set the processing-time timeout duration as shown in this code snippet:

// In Scala
          timeoutDurationMs = lastTaskTimstampMs + periodIntervalMs - groupState.getCurrentProcessingTimeMs()