Using autoloader with multiple object types in load path

[emorgoch]

My data source is going to generate csv files for multiple objects all into the same directory that I need to load from. The files will have names in the format along the lines of <objecttype>_YYYY_MM_DD_guid.csv.gz. Each objecttype will have it's own schema, and the goal is to load those files into objecttype tables.

I'm using Autoloader with cloudfiles to ingest these files. The objective is to have the process by dynamic and not have a static set of objecttypes, being able to adjust to new objecttypes being added. In my prototype, I'm performing an initial dbutils.fs.ls to get a file list and parse our the unique objecttypes. Then I loop through each objecttype, passing them to autoloader and globbing the objecttype out. But I feel this won't scale well as the number of files increases.

Is there a better method I can use to ingest these files and have them directed to the correct tables?

[ShamenParis]

Hi @emorgoch ,

Autoloader is absolutely the best tool when we just need to ingest new files as they land in storage. However, the way you are currently doing it—listing the files with dbutils and passing that list into Autoloader—isn't going to be very efficient. It actually bypasses a lot of the built-in magic that makes Autoloader so fast!

If you really need to cherry-pick specific files using dbutils, a much better approach is to use batching combined with threading.

Here is the catch you have to watch out for: standard Python threading only runs on the driver node. It can't distribute the Python threads themselves to your worker nodes. If you aren't careful, the threading will just run locally on the driver while your powerful worker nodes sit completely idle!

To get the full leverage of both your driver and your workers, we can use the threads on the driver just to kick off the Spark jobs. When we do this (and tell Spark to share resources), the driver handles the concurrency, but Spark distributes the actual heavy lifting across all your worker nodes in parallel.

By combining batching and threading this way, you'll see a massive performance boost. You can refer to the code below to see exactly how to set it up:

import concurrent.futures

# 1. Tell Spark to use the FAIR scheduler so concurrent threads can share the worker nodes
spark.conf.set("spark.scheduler.mode", "FAIR")

# Let's say this is your massive list of files from dbutils
all_files = [f"s3://your-bucket/file_{i}.parquet" for i in range(100)]

# 2. Break the files into manageable batches (e.g., 10 files per batch)
batch_size = 10
file_batches = [all_files[i:i + batch_size] for i in range(0, len(all_files), batch_size)]

# 3. Define the function that Spark will run for each batch
def process_batch(file_list):
    # Passing the list of files directly to spark.read distributes the work to the workers!
    df = spark.read.format("parquet").load(file_list)
    
    # Do your transformations and write the data
    df.write.format("delta").mode("append").save("/target/table")
    return f"Processed a batch of {len(file_list)} files successfully."

# 4. Use threading on the driver to submit these batches in parallel
# (max_workers defines how many parallel jobs the driver will try to submit at once)
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    # Submit all the batches
    futures = [executor.submit(process_batch, batch) for batch in file_batches]
    
    # Wait for them to finish
    for future in concurrent.futures.as_completed(futures):
        print(future.result())

[adnan_alvee]

Use one Auto Loader streaming table per object type, generated from a governed configuration registry;  not dbutils.fs.ls. Each object type needs an independent schema and checkpoint. Auto Loader cannot use one CSV stream to infer unrelated schemas and dynamically choose target tables: CSV inference produces one global schema for the stream.

Prefer changing the landing layout to:

<landing>/object_type=customer/...
<landing>/object_type=order/...

If that is impossible, keep the shared directory and assign each stream a non-overlapping pathGlobFilter.

Maintain a small registry containing:

• Validated object type
• Expected CSV schema
• Target table
• Optional CSV parsing options

Lakeflow Spark Declarative Pipelines can read this registry during pipeline planning and generate multiple streaming tables programmatically. The definitions are evaluated serially, but the resulting flows can execute in parallel. New objects become active on the next pipeline update. The registry should remain additive because removing a generated dataset causes that dataset to be dropped from the pipeline target schema.

Each generated flow:

1. Reads the landing path with Auto Loader.
2. Selects only its filenames using pathGlobFilter.
3. Applies that object’s explicit schema.
4. Writes to its own streaming table with independently managed state.

Auto Loader supports pre-compressed CSV files, including gzip. Lakeflow manages checkpoint and schema locations automatically; standalone streams require unique durable locations for each workload.

For production discovery, enable managed file events. They share one notification queue per external location, require Unity Catalog and DBR 14.3 LTS or later, and should run at least every seven days to avoid falling back to directory listing.

Minimal implementation example

import re
from pyspark import pipelines as dp

SOURCE = "/Volumes/<source_catalog>/<source_schema>/<landing_volume>"

# Small governed registry:
# object_type STRING, schema_ddl STRING, enabled BOOLEAN
objects = (
    spark.table("<config_catalog>.<config_schema>.ingestion_objects")
         .where("enabled = true")
         .select("object_type", "schema_ddl")
         .collect()
)

def define_object_table(object_type: str, schema_ddl: str):
    # Prevent target-name or glob injection.
    if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_]*", object_type):
        raise ValueError(f"Invalid object_type: {object_type}")

    @DP.table(name=f"bronze_{object_type.lower()}")
    def ingest_object():
        return (
            spark.readStream
                 .format("cloudFiles")
                 .option("cloudFiles.format", "csv")
                 .option("header", "true")
                 .option(
                     "pathGlobFilter",
                     f"{object_type}_????_??_??_*.csv.gz"
                 )
                 .option("cloudFiles.useManagedFileEvents", "true")
                 .option("rescuedDataColumn", "_rescued_data")
                 .schema(schema_ddl)
                 .load(SOURCE)
                 .selectExpr(
                     "*",
                     "_metadata.file_path AS _source_file",
                     "current_timestamp() AS _ingested_at"
                 )
        )

for row in objects:
    define_object_table(row.object_type, row.schema_ddl)

Another alternative could be to look into DLT-Meta, which is a datbaricks labs project. Its a meta data driven pipeline config for spark declarative pipelines. https://docs.databricks.com/aws/en/ldp/developer/dlt-meta

[emorgoch]

Thanks Adnan,

What I'm seeing with your solution, however, is that I need to maintain a static list of object types that I will be importing. This is what I am attempting to avoid, as system owners may expand the object types exported from the source to be ingested into Databricks without properly informing the data team.

[comb8342 · Accepted solution]

I’d avoid doing a dbutils.fs.ls followed by one Auto Loader stream per object type. That creates more streams/checkpoints as the number of object types grows.

A cleaner pattern is to use one Auto Loader stream over the directory, extract objecttype from the filename, and route each micro-batch to the appropriate Delta table with foreachBatch. Auto Loader is designed to scale to very large file counts, and the discovered file metadata is persisted in its checkpoint.

Conceptually:

df = (
spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("cloudFiles.schemaLocation", schema_path)
.option("header", "true")
.load(source_path)
)

df = df.withColumn(
"objecttype",
regexp_extract(input_file_name(),
r"/([^/_]+)_\d{4}_\d{2}_\d{2}_", 1)
)

def route_batch(batch_df, batch_id):
for obj in [r.objecttype for r in batch_df.select("objecttype").distinct().collect()]:
(
batch_df.filter(col("objecttype") == obj)
.drop("objecttype")
.write
.mode("append")
.saveAsTable(f"target.{obj}")
)

df.writeStream.foreachBatch(route_batch)...

The important part is that you’re not repeatedly scanning the directory to discover files. Auto Loader handles incremental discovery, while your routing logic determines the destination table.

One caveat: because each object type has its own schema, I’d maintain a small schema/metadata registry for object type → target table/schema rather than relying entirely on inference. Auto Loader supports schema inference/evolution, but different schemas going into different tables still need deliberate handling.

That architecture should scale much better than creating a separate Auto Loader stream for every object type. stickmanhook.com.br can be a separate reference while working on the pipeline.

[emorgoch]

Thanks comb for the suggestion and the sample code. Looking into the foreachbatch processing, this seems like if could work. My one concern with it is the information here about foreachbatch serializing the processing and removing parallelization, but this would probably be acceptable given the expected data sizes of the batched exports (a few dozen files, under 100000 records a batch).

[cartergray70543]

Use a single Auto Loader stream and extract the objecttype from _metadata.file_name. Then route each batch to the appropriate table based on that value. This scales much better than repeatedly listing files and creating separate streams for each object type.

[bijilsubhash]

I am curious - both @adnan_alvee and @ShamenParis has slightly different approach and both makes sense. Perhaps an open question - is the former more idiomatic even though both could work in this case?

[aayush_410]

The core issue with your prototype is that the file discovery step is being duplicated: Auto Loader already does incremental, stateful file discovery internally (and can use cloud-native file notifications instead of directory listing), but your dbutils.fs.ls loop is a separate manual full listing that doesn't scale and has to be re-run to find new objecttypes. The fix is to stop doing objecttype discovery before the stream and instead do it inside a single stream.

Recommended pattern: one Auto Loader stream landing everything, fan-out in foreachBatch

One Auto Loader stream reads the whole directory, not per-objecttype:
python
raw = (spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("cloudFiles.useNotifications", "true") # avoid repeated full listing as volume grows
.option("cloudFiles.schemaLocation", schema_loc)
.option("header", "true")
.schema(generic_string_schema) # or land everything as strings — see below
.load(source_path)
)

Since each objecttype has a different real schema, don't try to get Auto Loader to infer one unified schema across all of them — that's what was pushing you toward per-objecttype streams in the first place. Instead, either:

Land every column as string (schema inference will do this by default for CSV if you don't force it otherwise) plus _rescued_data, or
Land the whole row as a single VARIANT column — Databricks explicitly supports this pattern for exactly this kind of "many shapes landing in one place" ingestion, and it's schema-agnostic by construction.
Extract objecttype from the filename inside the stream, using the built-in _metadata column rather than a separate listing pass:
python
from pyspark.sql.functions import regexp_extract, col

raw = raw.withColumn(
"objecttype",
regexp_extract(col("_metadata.file_name"), r"^([^_]+)_", 1)
)

This means new objecttypes are picked up automatically the moment their files land — no code change, no stream restart, no separate scan.

Fan out to per-objecttype tables in foreachBatch, dynamically, based on whatever objecttypes actually appear in each micro-batch:
python
def route_batch(batch_df, batch_id):
for obj_type in [r.objecttype for r in batch_df.select("objecttype").distinct().collect()]:
subset = batch_df.filter(col("objecttype") == obj_type)
target_table = f"bronze.{obj_type}"
(subset.write.format("delta")
.mode("append")
.option("mergeSchema", "true")
.saveAsTable(target_table)) # creates the table on first appearance

raw.writeStream.foreachBatch(route_batch).option("checkpointLocation", checkpoint_loc).start()

Why this scales where your prototype doesn't:

One stream, one checkpoint — no per-objecttype glob patterns to maintain or restart as new types appear.
No separate full-directory listing to enumerate objecttypes — that work is now done incrementally by Auto Loader's own file-tracking state, and cheaply, per micro-batch, only on files that already got pulled in.
New objecttypes just work — the first time a new <objecttype>_*.csv.gz lands, foreachBatch sees a new distinct value and creates the table, no code deploy required.

One thing to decide up front: if downstream consumers need real typed columns (not everything as strings/VARIANT), you'll want a lightweight per-objecttype schema registry (even just a small control Delta table: objecttype -> expected_schema_json) that the foreachBatch function looks up to cast columns properly before writing — otherwise you're pushing the "what's the real schema" problem one layer downstream instead of solving it. That's a reasonable v2; landing as strings/VARIANT + rescue column first, then adding typed casting once you've stabilized the objecttype list, keeps the initial rollout simple.

[stephen4]

A single Auto Loader stream with filename-based routing looks much cleaner here, especially since new object types can appear without updating the pipeline.