Streaming to Delta Lake with Checkpointing

The Anatomy of a Streaming Write

Just as spark.readStream is the entry point for reading a stream, DataFrame.writeStream is the starting point for writing one. It allows you to define the sink, the output mode, and other crucial configurations for your stream.

The general syntax pattern in PySpark is:

streaming_query = (streaming_df
  .writeStream
  .format("delta")
  .outputMode("append")
  .option("checkpointLocation", "/path/to/checkpoint")
  .toTable("my_target_table")
)

Let's break down these components:

• .writeStream: This signals that you are defining a streaming sink for your DataFrame.
• .format("delta"): Specifies that you're writing to a Delta table.
• .outputMode("append"): This is one of the most important settings. append mode means that only new rows processed since the last trigger will be added to the sink table. It's the default and most common mode for simple ETL streams. We'll cover other modes in a future lesson.
• .option("checkpointLocation", ...): This is a mandatory option for most production-grade streams. It specifies a path where Spark will store the progress of the stream.
• .toTable("..."): This is a convenient method to start the stream and write the output to a table registered in the Unity Catalog or Hive Metastore. You can also use .start("/path/to/table") to write to a specific file path.

The Heart of Reliability: Checkpointing

Before we write any code, we must understand checkpointing. It is the single most important concept for ensuring your streaming queries are reliable and fault-tolerant.

A checkpoint is a directory on cloud storage where Spark records the state of your stream. This state includes:

1. Offsets: A record of which data has been read from the source (e.g., which files have been processed, what was the last offset read from Kafka).
2. Stateful Operator Data: For more complex streams with aggregations, this is where the intermediate state is stored.

This checkpoint acts as the stream's memory.

Why is this so critical?

• Fault Tolerance: If your stream stops for any reason (cluster failure, manual stop, etc.), the checkpoint allows it to restart from exactly where it left off, without missing any data or reprocessing old data.
• Exactly-Once Guarantee: By meticulously tracking what's been processed, checkpointing, combined with a transactional sink like Delta Lake, provides an end-to-end exactly-once processing guarantee.

The following video explains the concept and the syntax for checkpointLocation.

A crucial rule to remember is: Every streaming query must have its own unique checkpoint directory. Sharing a checkpoint directory between two different streams will lead to incorrect results and failures.

Writing a Stream to a Delta Sink

Now that we understand the importance of checkpointing, let's see the code in action. The official Databricks documentation provides a clear, minimal example of the syntax.

This next video provides a great walkthrough of a full read-and-write stream, demonstrating how the pieces fit together.

Your Turn: Build a Complete Streaming Pipeline

It's time for you to write the code. We'll build on the Auto Loader example from our last lesson to create a complete pipeline that reads from files and writes to a Delta table.

Goal: Create a streaming pipeline that reads CSV files with Auto Loader and writes the data to a Delta table.

1. Set Up Your Notebook: In a new Databricks notebook, run the following code to prepare a source directory and drop a sample CSV file into it.

   
   
   
   # Define paths for the source data, the target table, and the checkpoint
   source_path = "/FileStore/streaming_sink_input"
   table_name = "streaming_output_table"
   checkpoint_path = f"/tmp/delta/{table_name}/_checkpoint"
   
   
   
   
   # Clean up previous runs
   dbutils.fs.rm(source_path, recurse=True)
   dbutils.fs.rm(checkpoint_path, recurse=True)
   spark.sql(f"DROP TABLE IF EXISTS {table_name}")
   
   
   
   
   # Create the source directory
   dbutils.fs.mkdirs(source_path)
   
   
   
   
   # Create a sample CSV file
   csv_data = """
   transaction_id,item,amount
   1,ItemA,19.99
   2,ItemB,25.00
   """
   dbutils.fs.put(f"{source_path}/data_1.csv", csv_data, overwrite=True)
   print("Setup complete. A sample CSV file has been created.")
   

2. Define the Streaming Read: This is the readStream code from our previous lesson.

   
   
   
   # Configure the Auto Loader stream to read the CSV files
   streaming_df = (spark.readStream
     .format("cloudFiles")
     .option("cloudFiles.format", "csv")
     .option("header", "true")
     .option("cloudFiles.inferColumnTypes", "true")
     .option("cloudFiles.schemaLocation", checkpoint_path) # Auto Loader uses the same path for schema and source tracking
     .load(source_path)
   )
   

3. Define and Start the Streaming Write: Now, it's your turn. Complete the code below to write streaming_df to the table_name defined in step 1. You must specify the output mode and the checkpoint location.

   
   
   
   # YOUR CODE HERE
   # Complete the writeStream operation
   streaming_query = (streaming_df
     .writeStream
     .format("delta")
     .outputMode("append") # Use append mode to add new records
     .option("checkpointLocation", checkpoint_path) # Use the same checkpoint path
     .toTable(table_name)
   )
   

   Hint: You need to chain methods like .format(), .outputMode(), .option(), and .toTable().

4. Verify the Output: Once the stream is running, open a new cell and query your target table. You should see the two records from data_1.csv.

   -- In a new SQL cell
   SELECT * FROM streaming_output_table;
   

5. Test the End-to-End Stream: While the stream is still active, run the following code in another cell to drop a new file into the source directory.

   
   
   
   # Add a new file to the source directory
   new_csv_data = """
   transaction_id,item,amount
   3,ItemC,10.50
   4,ItemD,55.10
   """
   dbutils.fs.put(f"{source_path}/data_2.csv", new_csv_data, overwrite=True)
   

   Now, re-run the SELECT query from step 4. You should see the two new rows appear in your table automatically. Congratulations, you've built a working, end-to-end streaming pipeline!

Understanding Checkpoint Failure and Recovery

To truly appreciate the power of checkpointing, let's see what happens when it's gone.

After watching the video, you can stop the streaming query in your notebook (click the "Cancel" button on the cell). If you were to re-run the cell containing the writeStream code, it would pick up from where it left off. But if you were to delete the checkpoint_path directory first and then restart, it would re-process data_1.csv and data_2.csv, adding duplicate rows to your streaming_output_table. This is why protecting your checkpoint directory is so important.

Knowledge Check

Let's test your understanding with a couple of quick questions.

Question 1:
You are configuring a streaming write. Which option is used to specify the directory for storing the stream's state and progress?
A. checkpointDirectory
B. checkpointPath
C. checkpointLocation
D. checkpoint

Question 2:
You are building a pipeline to ingest files from cloud storage. To ensure that files are processed exactly once, even if the stream restarts after a failure, which of the following is most critical?
A. Using .outputMode("append").
B. Setting .trigger(availableNow=True).
C. Specifying a unique and persistent checkpointLocation.
D. Using .format("delta") for the sink.

Conclusion

In this lesson, you have completed the fundamental workflow of a Structured Streaming job. You now have the skills to not only read streaming data but also to write it reliably to a Delta table, the foundation of the Lakehouse architecture.

Key Takeaways

• You use DataFrame.writeStream to write a streaming DataFrame to a sink.
• The format("delta") specifies that the sink is a Delta table, which provides transactional guarantees.
• Checkpointing is mandatory for reliable streaming. You must provide a unique path using the .option("checkpointLocation", ...) for each stream.
• The checkpoint allows a stream to recover from failures and ensures exactly-once processing by remembering what data it has already processed.
• The .toTable() method is a convenient way to start a stream and register its output as a table in the metastore.

Next Steps

We've used the default append output mode in this lesson. However, different use cases require different behaviors. For example, how would you write the results of a streaming aggregation where the entire result needs to be updated with each batch?

In our next lesson, we will explore this by learning to select the appropriate output mode (append, complete, update) for a given streaming scenario.