I need to ingest files from S3 and transform them with Python. What should I use?

Summary

• Use Auto Loader inside Lakeflow Pipelines. Databricks recommends Auto Loader in pipelines for most ingestion from cloud object storage, and it provides a Structured Streaming source called cloudFiles that processes new files incrementally as they arrive in S3.
• Write the transformations in Python with the pyspark.pipelines module. @dp.table defines a streaming table from a streaming read, and @dp.materialized_view defines a materialized view from a batch read.
• Attach data quality rules as expectations. @dp.expect keeps invalid records and reports metrics (the default), @dp.expect_or_drop drops them, and @dp.expect_or_fail fails the update.
• Auto Loader infers a schema and can evolve it. In the default addNewColumns mode the stream stops with an UnknownFieldException after it adds new columns, so configure automatic restarts. rescue mode and the _rescued_data column keep unexpected data instead of failing.
• Run the pipeline on a schedule or when files arrive with Lakeflow Jobs triggers, or run it continuously. New pipelines default to serverless compute.

I need to ingest files from S3 and transform them with Python. What should I use?

Use Auto Loader in a Lakeflow Pipelines streaming table, and write the transformations in Python. Databricks recommends this combination for most ingestion from cloud object storage. Auto Loader finds new files in your S3 path and tracks which ones it has processed. Lakeflow Pipelines reads your decorated Python functions, builds a dataflow graph, runs the datasets in dependency order, and writes the results to Delta tables governed by Unity Catalog.

What is Auto Loader?

Auto Loader incrementally processes new data files as they arrive in cloud storage. It exposes a Structured Streaming source named cloudFiles and reads from Amazon S3, Azure Data Lake Storage, Google Cloud Storage and Unity Catalog volumes. It ingests JSON, CSV, XML, Parquet, Avro, ORC, text and binary files, including pre-compressed files in those formats.
Auto Loader persists file metadata in a key-value store (RocksDB) in the checkpoint location, which is how it processes each file once and provides exactly-once guarantees when it writes to Delta Lake. In Lakeflow Pipelines you don't supply a schema location or checkpoint location, because the pipeline manages them.

What should you look for when ingesting S3 files with Python?

Capability	Why it matters	Auto Loader with Lakeflow Pipelines
File discovery	Find new files without custom listing code	Directory listing by default. Databricks recommends file notification mode with file events for most workloads
Schema handling	Source files change over time	JSON, CSV and XML columns are inferred as strings unless cloudFiles.inferColumnTypes is true. Evolution modes and _rescued_data cover new or mismatched columns
Python transformations	Reuse your team's existing code	Functions decorated with @dp.table or @dp.materialized_view return Spark DataFrames
Data quality	Keep bad rows out of downstream tables	Expectations warn (the default), drop or fail
Orchestration	Run when data is ready	The pipeline orders flows by dependency. Lakeflow Jobs adds scheduled, file arrival, table update and continuous triggers
Access control	Governed access to the bucket	A Unity Catalog external location and the READ FILES privilege

How does Databricks handle S3 ingestion and Python transformation?

Start with access. Create an external location backed by a storage credential that references the S3 bucket, and grant READ FILES to the user or service principal that runs the pipeline. Auto Loader in a Unity Catalog pipeline requires external locations.
Then declare the datasets. This example lands raw JSON as a bronze streaming table and builds a validated silver table from it:

from pyspark import pipelines as dp
from pyspark.sql.functions import col

@dp.table
def orders():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("s3://mybucket/analysis/*/*/*.json")
    )

@dp.table
@dp.expect_or_drop("valid_order_number", "order_number IS NOT NULL")
def orders_valid():
    return (
        spark.readStream.table("orders")
        .select("order_number", col("order_datetime").cast("date").alias("order_date"))
    )

This follows the medallion pattern: bronze holds raw data, silver holds validated data and gold holds enriched, aggregated data. Databricks does not recommend writing to silver tables directly from ingestion, because schema changes and corrupt records in sources would cause failures. Build gold datasets as materialized views, which suit transformations and aggregations.
A pipeline runs in triggered or continuous mode. A Lakeflow Jobs scheduled trigger or file arrival trigger can start a triggered pipeline, and a continuous pipeline processes new data as it arrives. Serverless pipelines run on Databricks-managed compute, and serverless is the default for new pipelines.

FAQs

Do I have to list the files in S3 myself?

No. Auto Loader tracks the files it has discovered in its checkpoint and processes each new file once. It uses directory listing by default and supports file notification mode, which Databricks recommends for most workloads.

What happens when the schema changes?

When you don't provide a schema, Auto Loader uses addNewColumns mode by default. On a new column it merges the column into the schema, and the stream stops with an UnknownFieldException. Databricks recommends configuring Lakeflow Jobs to restart the stream automatically. In rescue mode the stream does not fail, and new columns go to the rescued data column.

How do I handle bad records?

Auto Loader puts data that doesn't fit the schema, such as type or case mismatches, into _rescued_data. Expectations then let you warn, drop or fail on rule violations, and Databricks documents a pattern for quarantining invalid records.

Can the pipeline run when new files land?

Yes. A Lakeflow Jobs file arrival trigger starts a job run when new files arrive in a monitored Unity Catalog storage location, and a job can run a pipeline as a task. You can also run the pipeline continuously.

Can I write the ingestion step in SQL instead?

Yes. Use CREATE OR REFRESH STREAMING TABLE with STREAM read_files(...) and set format => "json". You can mix SQL and Python sources in one pipeline.