Incrementally ingest files with Auto Loader

Files keep landing. A partner drops a new batch of order JSON into a volume every few minutes, and your job is to pick up each new file exactly once, cope with the schema drifting over time, and never let one malformed record stall the stream. That is what Databricks Auto Loader does — and it is the primary way to get files into the lakehouse.

You could hand-write the reader — spark.readStream.format("cloudFiles") with a stack of .option(...) calls for the schema location, inference, evolution, and rescued-data column, then a create_streaming_table and an append_flow to land the rows. Or you declare a type: cloudfiles load and let Lakehouse Plumber write that Lakeflow for you. That is the idea on every page: declare your ETL, don’t hand-write it.

Let’s ingest a landing folder of order JSON files into a bronze streaming table — incrementally, with the schema inferred and allowed to evolve — without touching PySpark.

Before you start

You need Lakehouse Plumber installed and a project to work in. If you followed the Get Started course you already have one; otherwise scaffold an empty project first. This guide assumes order files are already arriving under a landing path — a Unity Catalog volume, or an S3, ADLS, or GCS location. Auto Loader reads files that land there; it does not create them.

Declare the Auto Loader load

A pipeline in Lakehouse Plumber is a flowgroup: a short YAML file describing a sequence of actions. This one has two — a load that streams the landing folder into a view, and a write that appends the view into a streaming table.

Create pipelines/orders_ingest.yaml:

pipelines/orders_ingest.yaml
# Ingest guide: incrementally pick up new order files with Auto Loader.
# Action chain: load (cloudFiles JSON, schema inference + evolution) -> write (streaming_table).
pipeline: bronze_ingest
flowgroup: orders_ingest

actions:
  # 1. Load: continuously stream new JSON files from the landing volume with
  #    Auto Loader. Infer the schema, evolve it as new fields appear, and
  #    capture anything that fails type coercion in a rescued-data column.
  - name: load_orders_files
    type: load
    source:
      type: cloudfiles
      path: "${landing_path}/orders/*.json"
      format: json
      options:
        cloudFiles.schemaLocation: "${schema_location}"
        cloudFiles.inferColumnTypes: true
        cloudFiles.schemaEvolutionMode: addNewColumns
        cloudFiles.rescuedDataColumn: _rescued_data
    target: v_orders_raw

  # 2. Write: append every new file's rows into a bronze streaming table.
  - name: write_orders_bronze
    type: write
    source: v_orders_raw
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${bronze_schema}"
      table: orders_raw

The load action names the source with source.type: cloudfiles, points path at a glob of the landing folder, and sets format: json. Auto Loader also reads csv, parquet, avro, orc, text, and binaryFile; json is the default. You write format once — Lakehouse Plumber copies it into the mandatory cloudFiles.format reader option at generation time, so you never repeat yourself.

Auto Loader is a streaming source, so there is no batch alternative: LHP defaults readMode to stream and rejects batch. That is why the action has no readMode line — the default is the only valid value.

The four cloudFiles.* options under options are the schema story:

  • cloudFiles.schemaLocation — the durable location where Auto Loader persists the schema it infers and tracks how it changes across runs.

  • cloudFiles.inferColumnTypes: true — infer real types (numbers, timestamps) from the JSON instead of reading every field as a string.

  • cloudFiles.schemaEvolutionMode: addNewColumns — when a field appears in a later file that wasn’t in the inferred schema, add it to the table rather than failing.

  • cloudFiles.rescuedDataColumn: _rescued_data — capture any value that doesn’t fit its inferred type in a sidecar JSON column, so a malformed record neither fails the stream nor silently disappears.

The ${...} tokens are placeholders you resolve per environment, so the same flowgroup runs unchanged in dev, staging, and prod.

Point the tokens at your environment

Those tokens resolve from a per-environment substitutions file. Create substitutions/dev.yaml:

substitutions/dev.yaml
# Development environment tokens.
# Referenced from flowgroups with ${token} syntax.
dev:
  catalog: dev_catalog
  bronze_schema: bronze
  landing_path: /Volumes/dev_catalog/landing/files
  schema_location: /Volumes/dev_catalog/landing/_schemas/orders

This is the only place environment-specific values live. To ingest from a production landing zone later, you write a prod.yaml with the production catalog, landing path, and schema location — the flowgroup itself never changes.

Generate the pipeline

Now compile the YAML into Lakeflow code. Validate first, then generate:

$ lhp validate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze_ingest  0 files
✓ validate (0.32s)
1 validated · 0.3s

$ lhp generate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze_ingest  1 file
✓ generate (0.36s)
✓ format (0.05s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.4s

validate resolves every token and checks the actions before you commit to generating; generate writes the Python.

Read what Lakehouse Plumber wrote

Open generated/dev/bronze_ingest/orders_ingest.py. This is the entire output — nothing is hidden behind a runtime:

generated/dev/bronze_ingest/orders_ingest.py
# Generated by LakehousePlumber
# Pipeline: bronze_ingest
# FlowGroup: orders_ingest

from pyspark import pipelines as dp

# Pipeline Configuration
PIPELINE_ID = "bronze_ingest"
FLOWGROUP_ID = "orders_ingest"


# ============================================================================
# SOURCE VIEWS
# ============================================================================


@dp.temporary_view()
def v_orders_raw():
    """Load data from json files at /Volumes/dev_catalog/landing/files/orders/*.json"""
    df = (
        spark.readStream.format("cloudFiles")
        .option(
            "cloudFiles.schemaLocation", "/Volumes/dev_catalog/landing/_schemas/orders"
        )
        .option("cloudFiles.inferColumnTypes", True)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("cloudFiles.rescuedDataColumn", "_rescued_data")
        .option("cloudFiles.format", "json")
        .load("/Volumes/dev_catalog/landing/files/orders/*.json")
    )

    return df


# ============================================================================
# TARGET TABLES
# ============================================================================

# Create the streaming table
dp.create_streaming_table(
    name="dev_catalog.bronze.orders_raw", comment="Streaming table: orders_raw"
)


# Define append flow(s)
@dp.append_flow(
    target="dev_catalog.bronze.orders_raw",
    name="f_orders_bronze",
    comment="Append flow to dev_catalog.bronze.orders_raw",
)
def f_orders_bronze():
    """Append flow to dev_catalog.bronze.orders_raw"""
    # Streaming flow
    df = spark.readStream.table("v_orders_raw")

    return df

It’s ordinary Lakeflow. The source view is a @dp.temporary_view wrapping spark.readStream.format("cloudFiles"), with each .option(...) you declared emitted in order — the boolean inferColumnTypes as a bare True, the rest as strings — and cloudFiles.format injected from source.format. The target is a dp.create_streaming_table, and a @dp.append_flow streams the view into dev_catalog.bronze.orders_raw. You described a landing folder and a target table; LHP wrote the reader and the two pieces of plumbing that connect them.

Because Auto Loader records which files it has already ingested in the checkpoint beside schemaLocation, each run of the pipeline processes only files that are new since the last run. That is the incremental behaviour you came for — you get it from the cloudfiles type, not from any code you wrote.

Choose a schema strategy

The load above lets Auto Loader infer and evolve the schema — the right default when JSON files drift over time and you’d rather absorb new fields than break. Two other strategies trade discovery for control, and you pick per load:

  • Guide the inference with cloudFiles.schemaHints. Pin the types you care about — "order_id BIGINT, amount DOUBLE" — while Auto Loader still discovers the rest. The hint accepts an inline DDL string, a YAML schema file (where a column marked nullable: false becomes NOT NULL), or a .ddl / .sql file; LHP picks the form by file extension.

  • Enforce a fixed schema with source.schema. Point it at a schema file and Auto Loader accepts exactly those columns and turns inference off — so schema evolution no longer applies.

source.schema and cloudFiles.schemaHints are two ways to say different things, and you set at most one. The load action reference documents the schema file formats and the exact rendering for each.

Note

Declaring more than one schema source on a single load — source.schema, cloudFiles.schemaHints, or the legacy schema_file — is a configuration conflict, and validate rejects it before generation. Pick one strategy per load.

Control which files enter the stream

source.path is passed straight into the generated .load(...) with no rewriting, so any glob Spark and Auto Loader support works as-is. Narrow it to scope the directories Auto Loader lists: brace alternatives like {sales,marketing,events} scan only those folders, character classes like [0-9] match ranges, and [^x] negates a set. Use [^x], not the shell’s [!x], and avoid the ** globstar — it is undocumented for Auto Loader. On DBR 12.2+, set cloudFiles.useStrictGlobber: "true" for predictable, Spark-standard globbing.

To filter on the file name rather than the path, set pathGlobFilter: "*.parquet" under options — it matches each file’s basename after listing. For a rule no glob can express, filter downstream in a SQL transform on the _metadata.file_path column that Auto Loader exposes at read time. The load action reference covers the full glob and path-filter set.

Fan several sources into one table

When the same order schema arrives from several landing zones — one bucket per region, say — consolidate them into one bronze table without merging the streams. Declare one load + write pair per source, and point every write at the same streaming_table. The first write keeps the default create_table: true and creates the table; each later write sets create_table: false and appends to it. LHP emits a single dp.create_streaming_table and one @dp.append_flow per source, each with its own checkpoint — so one slow or broken source doesn’t stall the others. The write action reference documents the streaming-table and append-flow rules in full.

What you just did

Twenty-three lines of YAML. Fifty-six lines of Lakeflow Python. Zero lines of PySpark from you — you never wrote readStream.format("cloudFiles"), the five-option reader chain, create_streaming_table, or append_flow. And the output is code you own: version it, diff it, and open it in the Databricks editor like anything else.

The payoff compounds. Adding schema hints, narrowing the glob, or fanning in a second landing zone is a few more lines of YAML — not another fifty-six lines of Python each time.

What’s next

  • Handle the rescued rows. Pair cloudFiles.rescuedDataColumn with a data-quality gate to drop or quarantine records that landed malformed, so only clean rows reach downstream tables. See the data-quality guide.

  • Tune throughput and listing. cloudFiles.maxFilesPerTrigger and cloudFiles.maxBytesPerTrigger cap each micro-batch; cloudFiles.useIncrementalListing and cloudFiles.useNotifications change how Auto Loader discovers new files on large directories. The load action reference lists them with their constraints.

  • Reprocess files. cloudFiles.includeExistingFiles (first run only), cloudFiles.allowOverwrites, and cloudFiles.cleanSource control whether already-seen files are re-ingested or archived.

  • Add file-lineage columns. operational_metadata attaches audit columns such as _source_file_path, _source_file_size, and _source_file_modification_time to every row.

  • See every option. The full cloudFiles.* set, with defaults and constraints, is in the load action reference.