Incremental Data Ingestion with Auto Loader

1. The Challenge of Incremental Ingestion

Before diving into the solution, it's important to understand the problem. In most data platforms, you receive new data files continuously in cloud storage (like AWS S3, Azure Data Lake Storage, or GCP Cloud Storage). The goal is to process only the new files as they arrive, without reprocessing the entire dataset.

This sounds simple, but it comes with several challenges:

• How do you keep track of which files have already been processed?
• How do you handle a very large number of files or deeply nested directory structures efficiently?
• How do you ensure that each file is processed exactly once, even if the pipeline fails and restarts?

To frame these challenges, start by watching the beginning of this video.

Auto Loader is Databricks' answer to these challenges. It's a purpose-built service for building simple, scalable, and robust incremental ingestion pipelines.

At its core, Auto Loader works by monitoring a source directory in cloud storage. It maintains the state of which files have been processed in a checkpoint location, using a highly efficient embedded database (RocksDB). This allows it to quickly identify new files and process them incrementally.

Auto Loader has two modes for detecting new files:

1. Directory Listing Mode (Default): Periodically lists the source directory to find new files. It's highly optimized and works out-of-the-box without extra permissions.
2. File Notification Mode: Uses cloud-native event notification services (e.g., AWS SQS, Azure Queue Storage) to get notified when a new file arrives. This is more scalable for directories with millions of files but requires additional setup.

2. Configuring Your First Auto Loader Stream

Let's get hands-on. Configuring an Auto Loader job involves using the Spark Structured Streaming API. Since you're proficient with Python and Spark, the syntax will look familiar. The key is using cloudFiles as the format and setting a few critical options.

The video below provides a clear, practical demonstration of setting up an Auto Loader stream from scratch.

The Core Syntax

As you saw in the video, the basic structure of an Auto Loader job in PySpark is:

1. Read Stream: Define the source using spark.readStream.
2. Write Stream: Define the sink (your target Delta table) using .writeStream.

Here is a breakdown of the most critical components:

# 1. Define variables for your paths
input_path = "/path/to/your/source/data/"
schema_location = "/path/to/your/schema/location/"
checkpoint_location = "/path/to/your/checkpoint/location/"
target_table = "my_bronze_table"




# 2. Configure the Read Stream
df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")  # Specify the format of your source files
  .option("cloudFiles.schemaLocation", schema_location) # Auto Loader tracks schema here
  .load(input_path) # The directory to monitor
)




# 3. Configure the Write Stream
(df.writeStream
  .option("checkpointLocation", checkpoint_location) # Auto Loader tracks processed files here
  .option("mergeSchema", "true") # Allow schema evolution in the target table
  .trigger(availableNow=True) # Process all available data and then stop (like a batch job)
  .toTable(target_table) # Write to a Delta table
)

Two Critical Locations:

• cloudFiles.schemaLocation: Auto Loader automatically infers the schema of your data. It stores this inferred schema and any future changes (schema evolution) in this location. This makes your pipeline resilient to schema changes in the source data.
• checkpointLocation: This is the heart of incremental processing and fault tolerance. Structured Streaming records which files have been successfully processed in this directory. If the job stops and restarts, it uses the checkpoint to resume from where it left off, guaranteeing exactly-once processing.

Hands-On Lab: Ingesting JSON data

Now it's your turn. In a Databricks notebook, adapt and run the following code. For this exercise, you can use a directory on DBFS or an external cloud storage location you have access to.

Step 1: Prepare Your Data and Paths
First, create some sample JSON files and place them in a directory. For example, create a directory /FileStore/autoloader_input/ and upload two simple JSON files there.

• file1.json: {"id": 1, "name": "Alice", "city": "New York"}
• file2.json: {"id": 2, "name": "Bob", "city": "London"}

Step 2: Configure and Run the Auto Loader Stream
In a new notebook cell, define your paths and run the stream.

# Import necessary functions
from pyspark.sql.functions import col, current_timestamp




# 1. Define your paths. 
# Best practice: use a user-specific path to avoid conflicts.
# Replace <your_name> with your name or a unique identifier.
username = "your_name" 
input_path = "/FileStore/autoloader_input/"
checkpoint_path = f"/tmp/{username}/_checkpoints/autoloader_demo"
table_name = f"{username}_autoloader_target"




# Clean up previous runs for a fresh start (optional)
dbutils.fs.rm(checkpoint_path, recurse=True)
spark.sql(f"DROP TABLE IF EXISTS {table_name}")




# 2. Read the stream with Auto Loader
raw_df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", f"{checkpoint_path}/schema")
  .load(input_path)
)




# 3. Add some simple transformations (enrichment)
processed_df = (raw_df
  .withColumn("source_file", col("_metadata.file_path"))
  .withColumn("processing_time", current_timestamp())
)




# 4. Write the stream to a Delta table
(processed_df.writeStream
  .option("checkpointLocation", f"{checkpoint_path}/data")
  .option("mergeSchema", "true")
  .trigger(availableNow=True)
  .toTable(table_name)
)

Step 3: Verify the Result
Once the stream finishes, query your target table.

%sql
SELECT * FROM your_name_autoloader_target;

You should see the data from both file1.json and file2.json, along with the source_file and processing_time columns you added.

Step 4: Test Incremental Ingestion
Now, add a new file, file3.json, to your input directory.

• file3.json: {"id": 3, "name": "Charlie", "city": "Paris"}

Re-run the same Auto Loader code from Step 2. Then, query the table again. You will see that only the new record for "Charlie" has been added. The existing records for Alice and Bob were not reprocessed. This demonstrates the power of the checkpoint location!

3. Common and Practical Configuration Patterns

Your background in data engineering means you'll quickly move beyond basic setups. Here are some common configuration patterns from the official Databricks documentation that you'll find useful in real-world scenarios.

Let's summarize those key options:

• Filtering files with pathGlobFilter: If your source directory contains mixed file types (e.g., .csv, .json, .txt), you can tell Auto Loader to only process specific ones.

  .option("pathGlobFilter", "*.json")
  

• Guiding schema with schemaHints: Sometimes, Auto Loader might infer a column as a string when you know it should be a timestamp or integer. schemaHints lets you provide gentle corrections without defining the entire schema.

  
  
  
  # Tell Auto Loader that the 'event_time' column is a TIMESTAMP
  .option("cloudFiles.schemaHints", "event_time TIMESTAMP")
  

• Inferring nested data with inferColumnTypes: By default, for performance, Auto Loader may treat complex JSON fields as strings. To have it infer the nested structure, you can enable this option.

  .option("cloudFiles.inferColumnTypes", "true")
  

These options give you fine-grained control over your ingestion pipeline while retaining the simplicity and power of Auto Loader's automated features.

Conclusion

Congratulations! You have successfully configured your first incremental ingestion pipeline using Auto Loader. This is a foundational skill for any data engineer working in the Databricks ecosystem.

Key Takeaways:

• Auto Loader simplifies incremental data ingestion by automatically tracking and processing new files from cloud storage.
• It is configured using spark.readStream.format("cloudFiles").
• The checkpointLocation is essential for state management and ensuring exactly-once processing.
• The cloudFiles.schemaLocation enables robust schema inference and evolution, making pipelines resilient to changes.
• You can run Auto Loader in a batch-like mode using .trigger(availableNow=True), which is perfect for scheduled jobs.

Preview of the Next Lesson:
In this lesson, we used the default directory listing mode. While it's great for most scenarios, what happens when you have millions of files arriving per hour? For that, you might need file notification mode. In our next lesson, we will compare Auto Loader's directory listing mode vs. file notification mode, diving into their performance characteristics, costs, and use cases, so you can choose the right tool for any job.