Reading Data Streams with `spark.readStream`

The Anatomy of a Streaming Read

The entry point for all streaming queries is spark.readStream. While its batch counterpart, spark.read, reads a static dataset and terminates, spark.readStream creates a DataFrame that represents a potentially infinite stream of data.

The general syntax pattern looks like this in PySpark:

streaming_df = (spark.readStream
  .format("source_format")
  .option("key1", "value1")
  .option("key2", "value2")
  .load("/path/to/source")
)

Let's break this down:

• spark.readStream: This is the explicit instruction to Spark to create a streaming source.
• .format(...): Specifies the type of the source, such as "delta" for a Delta table or "cloudFiles" for Auto Loader.
• .option(...): Provides specific configurations for the chosen format, like schema information or rate limits.
• .load(...): Specifies the path to the source data. Alternatively, you can use .table(...) for sources registered in the metastore.

Now, let's see how to apply this to the two most common streaming sources you'll encounter in Databricks.

1. Ingesting Files with Auto Loader (cloudFiles)

When you need to incrementally ingest files landing in cloud storage (like S3, ADLS Gen2, or GCS), the best-practice tool in Databricks is Auto Loader. It is an optimized streaming source that efficiently discovers and processes new files as they arrive.

Under the hood, Auto Loader is implemented as a Structured Streaming source called cloudFiles. You use it by setting the format in your readStream call to "cloudFiles".

This first video provides a great overview and practical demonstration of setting up and using Auto Loader.

As you saw, the key to using Auto Loader is format("cloudFiles") and providing a few essential options. The most important of these is cloudFiles.schemaLocation. Auto Loader uses this location to store state information about which files have been processed, a mechanism known as checkpointing. This is how it tracks the stream's progress and ensures files are processed exactly once, even if the stream stops and restarts.

The official Databricks documentation provides another clean example.

Your Turn: Set up an Auto Loader Stream

Now it's time for some hands-on practice. In your Databricks workspace, create a new notebook and try the following.

Goal: Read a stream of CSV files using Auto Loader.

1. Prepare a Source Directory: Create a directory in your DBFS (or a volume) where you can drop files. For example, /FileStore/streaming_input.

2. Prepare Sample Data: Create a simple CSV file with a header and a few rows, and upload it to the directory you just created.
   Example data_1.csv:

   id,value
   1,100
   2,200
   

3. Write the readStream Code: In your notebook, write the PySpark code to read from this directory. You'll need to define paths for your source data and for the checkpoint.

   
   
   
   # Define paths for your source data and checkpoint
   source_path = "/FileStore/streaming_input"
   checkpoint_path = "/tmp/ss-tutorial/_checkpoint"
   
   
   
   
   # Configure the Auto Loader stream
   streaming_df = (spark.readStream
     .format("cloudFiles")
     .option("cloudFiles.format", "csv")
     .option("cloudFiles.inferColumnTypes", "true") # Optional: let Auto Loader infer data types
     .option("header", "true") # Our CSV has a header
     .option("cloudFiles.schemaLocation", checkpoint_path)
     .load(source_path)
   )
   
   
   
   
   # To see the stream in action, display it
   # Note: In a real pipeline, you would use .writeStream instead of .display
   display(streaming_df)
   

4. Test the Stream: Run the cell. You should see a streaming temporary table that shows the data from data_1.csv. Now, without stopping the stream, upload a second file (data_2.csv) to the same directory. You should see the new rows appear in the output automatically!

2. Streaming from a Delta Table

The other fundamental streaming source is a Delta table itself. This is the backbone of the medallion architecture, where you might have one stream writing to a Bronze table, and a second stream reading from that Bronze table to apply transformations and write to a Silver table.

Because Delta Lake maintains a transaction log of all changes, it is a perfect source for streaming. Spark can simply query the transaction log to discover new data that has been committed since the last micro-batch.

This video demonstrates the full cycle: creating a small Delta table, and then setting up a readStream to process it.

The syntax is very straightforward. You can specify the source Delta table either by its path or, if it's registered in the metastore, by its name.

Let's look at the official documentation for the precise syntax.

Your Turn: Stream from a Delta Table

Let's apply this in your notebook.

Goal: Create a Delta table and read from it as a stream.

1. Create a Source Delta Table:

   
   
   
   # Create a sample DataFrame
   data = [("a", 1), ("b", 2)]
   columns = ["key", "value"]
   source_df = spark.createDataFrame(data, columns)
   
   
   
   
   # Write it as a Delta table
   source_df.write.format("delta").mode("overwrite").saveAsTable("my_source_table")
   

2. Write the readStream Code: Now, configure a stream to read from this table.

   
   
   
   # Read from the Delta table as a stream
   streaming_df_delta = spark.readStream.table("my_source_table")
   
   
   
   
   # Display the stream
   display(streaming_df_delta)
   

3. Test the Stream: Run the display cell. You will see the initial two rows. Now, in a new cell, append more data to the source table and run it.

   
   
   
   # Append new data to the source table
   more_data = [("c", 3), ("d", 4)]
   more_df = spark.createDataFrame(more_data, columns)
   
   more_df.write.format("delta").mode("append").saveAsTable("my_source_table")
   

   Go back to your streaming display. You will see the new rows from more_df appear in the output.

3. Important Configuration Options

When you create a streaming read, you can provide options to control its behavior. While there are many, a few are particularly useful to know about.

• maxFilesPerTrigger: This is very useful for controlling costs and resource usage. It prevents your stream from being overwhelmed if a large number of files land at once.
• startingVersion / startingTimestamp: These are powerful for production scenarios. For example, if you need to restart a failed stream or backfill data from a certain point in time, you can use these options to avoid reprocessing the entire table history.

Conclusion

In this lesson, you've moved from theory to practice. You now know how to create a streaming DataFrame—the fundamental building block of a Structured Streaming application—from the two most important data sources in the Databricks ecosystem.

Key Takeaways

• You start a streaming query with spark.readStream.
• To ingest files incrementally from cloud storage, you use Auto Loader by setting .format("cloudFiles") and specifying a cloudFiles.schemaLocation.
• To stream changes from a Delta table, you set .format("delta") and use .load("/path/to/table") or .table("table_name").
• You can control the behavior of a streaming read using options like maxFilesPerTrigger to limit the batch size or startingVersion to process data from a specific point in time.

Next Steps

Reading data is the first half of the equation. The second half is writing it somewhere. In our next lesson, Write streaming query results to a Delta table sink using spark.writeStream with checkpointing, we will learn how to take the streaming DataFrames we created today and write their output to a destination table, completing our first end-to-end streaming pipeline.